Applying interpolate

[jramborger78]

Hi all.

Been at this pretty much all weekend and this morning. I cannot seem to get interpolate as shown in the analyses examples to work in my script. ChatGPT couldn't even get it to stop skipping the first empties. I also read that the interp1d version is different than the pandas version, and figured you guys knew what you were doing by creating your own function for this data. I have tried a few ways but this is the current version I am attempting. I figured since i want a CSV and you guys play with the h5 that i can try to convert to h5 convert back to slp and then write as csv but also feel it should be easier than this... Any ideas? Thanks.

import subprocess
import os
import sleap
from sleap.info.write_tracking_h5 import main as write_analysis
import pandas as pd
import numpy as np
from scipy.interpolate import interp1d
import h5py

# Step 1: Set Paths to various directories and files
# Models
centroid = "C:/Users/jramb/Documents/Sleap/models/1_centroid_model/training_config.json"
centered = "C:/Users/jramb/Documents/Sleap/models/1_centered_model/training_config.json"

# SLEAP Track Parameters
parameters = "--tracking.tracker simplemaxtracks --max_instances 1 --tracking.max_tracks 1 --tracking.similarity instance --tracking.match hungarian --tracking.track_window 5 --tracking.post_connect_single_breaks 0 --tracking.max_tracking 1 --verbosity rich --no-empty-frames"

# Video Directory
video_path = "C:/Users/jramb/Documents/Sleap/results/videos"

# Inference Result Location
inf_path = "C:/Users/jramb/Documents/Sleap/results/track"

# Location of Files for Analyses
analysis_path = "C:/Users/jramb/Documents/Sleap/results/analyses"

# Box Number and Subject CSV
box_mapping_path = "C:/Users/jramb/Documents/Sleap/results/Box_Subject.csv"
box_mapping_df = pd.read_csv(box_mapping_path)


# Step 2: Define interpolation function
def fill_missing(Y, kind="linear"):
    """Fills missing values independently along each dimension after the first."""

    # Store initial shape.
    initial_shape = Y.shape

    # Flatten after first dim.
    Y = Y.reshape((initial_shape[0], -1))

    # Interpolate along each slice.
    for i in range(Y.shape[-1]):
        y = Y[:, i]

        # Build interpolant.
        x = np.flatnonzero(~np.isnan(y))
        f = interp1d(x, y[x], kind=kind, fill_value=np.nan, bounds_error=False)

        # Fill missing
        xq = np.flatnonzero(np.isnan(y))
        y[xq] = f(xq)
        
        # Fill leading or trailing NaNs with the nearest non-NaN values
        mask = np.isnan(y)
        y[mask] = np.interp(np.flatnonzero(mask), np.flatnonzero(~mask), y[~mask])

        # Save slice
        Y[:, i] = y

    # Restore to initial shape.
    Y = Y.reshape(initial_shape)

    return Y




# Step 3: Tracking, Interpolating, and CSV File Creation
for video_file in os.listdir(video_path):
    if video_file.endswith(".mp4"):
        video_path_full = os.path.join(video_path, video_file)
        video_name_no_ext = os.path.splitext(video_file)[0]
        session_name = video_name_no_ext.split("_")[0]
        box_number = video_name_no_ext.split("_")[1]
        subject = box_mapping_df.loc[box_mapping_df["Box"] == box_number, "Subject"].values[0]

        print(f"Begin Tracking For {video_name_no_ext}")

        # Run sleap-track
        track_command = f"sleap-track -m \"{centroid}\" -m \"{centered}\" {parameters} -o \"{inf_path}/{session_name}_{subject}.slp\" \"{video_path_full}\""
        subprocess.run(track_command, shell=True)

        print(f"Tracking Completed For {video_name_no_ext}")
		
		# Convert to h5
        convert_command = f"sleap-convert --format analysis -o \"{inf_path}/{session_name}_{subject}.h5\" \"{inf_path}/{session_name}_{subject}.slp\""
        subprocess.run(convert_command, shell=True)

        # Read the h5 file
        h5_file_path = f"{inf_path}/{session_name}_{subject}.h5"
        with h5py.File(h5_file_path, "r") as f:
            dset_names = list(f.keys())
            locations = f["tracks"][:].T  # Transpose the data

            # Apply fill_missing function
            locations = fill_missing(locations)

            # Transpose back
            locations = locations.T

        # Save the updated data back to the h5 file
        with h5py.File(h5_file_path, "a") as f:
            f["tracks"][:] = locations # Transpose before saving

        # Convert back to slp
        convert_back_command = f"sleap-convert --format sleap \"{h5_file_path}\" -o \"{inf_path}/{session_name}_{subject}.slp\""
        subprocess.run(convert_back_command, shell=True)

        print(f"Processing Complete For {session_name}_{subject}")
		
		
        # Write analysis to CSV format
        labels = sleap.load_file(f"{inf_path}/{session_name}_{subject}.h5")
        csv_path = f"{analysis_path}/{session_name}_{subject}.csv"
        write_analysis(
            labels=labels,
            output_path=csv_path,
            all_frames=True,
            video=labels.video,
            csv=True,
        )

        print(f"CSV File Created For {session_name}_{subject}")

        # Post-Process of CSV File
        # Read with index_col=False to avoid issues with index columns
        df = pd.read_csv(csv_path, index_col=False)

        # Add Subject to Columns
        new_column_names = [f"{col}_{subject}" for col in df.columns]

        # Rename Columns directly in the existing DataFrame
        df.columns = new_column_names

        # Save File without including the index
        df.to_csv(csv_path, index=False)

        print(f"Processing Complete For {session_name}_{subject}")

[jramborger78]

Update:

I believed this worked, with a small adjustment to the script above of adding the --video to the end of the conversion back from h5 to slp after interpolating. It appears to have done what I had hoped which is:

1. Take in a folder of videos
2. utilize a csv with subject names and respective boxes for naming conventions
3. Apply tracking on each video
4. convert the slp file to h5
5. interpolate using function in Analysis guide to fill_missing
6. convert back to slp file
7. write_analysis as csv
8. Add subject name to column names and resave csv file

This is the current version below that worked in case others want to adapt it for their own or you guys find it useful:

import subprocess
import os
import sleap
from sleap.info.write_tracking_h5 import main as write_analysis
import pandas as pd
import numpy as np
from scipy.interpolate import interp1d
import h5py

# Step 1: Set Paths to various directories and files
# Models
centroid = "C:/Users/jramb/Documents/Sleap/models/1_centroid_model/training_config.json"
centered = "C:/Users/jramb/Documents/Sleap/models/1_centered_model/training_config.json"

# SLEAP Track Parameters
parameters = "--tracking.tracker simplemaxtracks --max_instances 1 --tracking.max_tracks 1 --tracking.similarity instance --tracking.match hungarian --tracking.track_window 5 --tracking.post_connect_single_breaks 0 --tracking.max_tracking 1 --verbosity rich --no-empty-frames"

# Video Directory
video_path = "C:/Users/jramb/Documents/Sleap/results/videos"

# Inference Result Location
inf_path = "C:/Users/jramb/Documents/Sleap/results/track"

# Location of Files for Analyses
analysis_path = "C:/Users/jramb/Documents/Sleap/results/analyses"

# Box Number and Subject CSV
box_mapping_path = "C:/Users/jramb/Documents/Sleap/results/Box_Subject.csv"
box_mapping_df = pd.read_csv(box_mapping_path)


# Step 2: Define interpolation function
def fill_missing(Y, kind="linear"):
    """Fills missing values independently along each dimension after the first."""

    # Store initial shape.
    initial_shape = Y.shape

    # Flatten after first dim.
    Y = Y.reshape((initial_shape[0], -1))

    # Interpolate along each slice.
    for i in range(Y.shape[-1]):
        y = Y[:, i]

        # Build interpolant.
        x = np.flatnonzero(~np.isnan(y))
        f = interp1d(x, y[x], kind=kind, fill_value=np.nan, bounds_error=False)

        # Fill missing
        xq = np.flatnonzero(np.isnan(y))
        y[xq] = f(xq)
        
        # Fill leading or trailing NaNs with the nearest non-NaN values
        mask = np.isnan(y)
        y[mask] = np.interp(np.flatnonzero(mask), np.flatnonzero(~mask), y[~mask])

        # Save slice
        Y[:, i] = y

    # Restore to initial shape.
    Y = Y.reshape(initial_shape)

    return Y




# Step 3: Tracking, Interpolating, and CSV File Creation
for video_file in os.listdir(video_path):
    if video_file.endswith(".mp4"):
        video_path_full = os.path.join(video_path, video_file)
        video_name_no_ext = os.path.splitext(video_file)[0]
        session_name = video_name_no_ext.split("_")[0]
        box_number = video_name_no_ext.split("_")[1]
        subject = box_mapping_df.loc[box_mapping_df["Box"] == box_number, "Subject"].values[0]

        print(f"Begin Tracking For {video_name_no_ext}")

        # Run sleap-track
        track_command = f"sleap-track -m \"{centroid}\" -m \"{centered}\" {parameters} -o \"{inf_path}/{session_name}_{subject}.slp\" \"{video_path_full}\""
        subprocess.run(track_command, shell=True)

        print(f"Tracking Completed For {video_name_no_ext}")
		
		# Convert to h5
        convert_command = f"sleap-convert --format analysis -o \"{inf_path}/{session_name}_{subject}.h5\" \"{inf_path}/{session_name}_{subject}.slp\""
        subprocess.run(convert_command, shell=True)

        # Read the h5 file
        h5_file_path = f"{inf_path}/{session_name}_{subject}.h5"
        with h5py.File(h5_file_path, "a") as f:
            dset_names = list(f.keys())
            locations = f["tracks"][:].T  # Transpose the data

            # Apply fill_missing function
            locations = fill_missing(locations)

            # Transpose back and save the updated data
            f["tracks"][:] = locations.T


        # Convert back to slp
        convert_back_command = f"sleap-convert --format sleap \"{h5_file_path}\" -o \"{inf_path}/{session_name}_{subject}.slp\" --video \"{video_path_full}\""
        subprocess.run(convert_back_command, shell=True)

        print(f"Processing Complete For {session_name}_{subject}")
		
		
        # Write analysis to CSV format
        labels = sleap.load_file(f"{inf_path}/{session_name}_{subject}.slp")
        csv_path = f"{analysis_path}/{session_name}_{subject}.csv"
        write_analysis(
            labels=labels,
            output_path=csv_path,
            all_frames=True,
            video=labels.video,
            csv=True,
        )

        print(f"CSV File Created For {session_name}_{subject}")

        # Post-Process of CSV File
        # Read with index_col=False to avoid issues with index columns
        df = pd.read_csv(csv_path, index_col=False)

        # Add Subject to Columns
        new_column_names = [f"{col}_{subject}" for col in df.columns]

        # Rename Columns directly in the existing DataFrame
        df.columns = new_column_names

        # Save File without including the index
        df.to_csv(csv_path, index=False)

        print(f"Processing Complete For {session_name}_{subject}")