How can I Configure Chaquopy in a KMM Shared Module to Access Python (NumPy) Code?

I'm developing a Kotlin Multiplatform app and need to use Python (with the numpy package) in the shared module for processing .pkl files.
I followed the official setup as described in the Official Setup Link-Kotlin.

My configuration files are as follows:

Project-setting.gradle.kts

KOTLINCopied
rootProject.name = "Assignment"
enableFeaturePreview("TYPESAFE_PROJECT_ACCESSORS")

pluginManagement {
    repositories {
        google {
            mavenContent {
                includeGroupAndSubgroups("androidx")
                includeGroupAndSubgroups("com.android")
                includeGroupAndSubgroups("com.google")
            }
        }

        mavenCentral()
        gradlePluginPortal()
    }
}

dependencyResolutionManagement {
    repositories {
        google {
            mavenContent {
                includeGroupAndSubgroups("androidx")
                includeGroupAndSubgroups("com.android")
                includeGroupAndSubgroups("com.google")
            }
        }
        mavenCentral()
    }
}

include(":composeApp")
include(":shared")

Project -> build.gradle.kts

KOTLINCopied
plugins {
    // this is necessary to avoid the plugins to be loaded multiple times
    // in each subproject's classloader
    alias(libs.plugins.androidApplication) apply false
    alias(libs.plugins.androidLibrary) apply false
    alias(libs.plugins.composeMultiplatform) apply false
    alias(libs.plugins.composeCompiler) apply false
    alias(libs.plugins.kotlinMultiplatform) apply false
    id("com.chaquo.python") version "16.0.0" apply false
}

Shared -> build.gradle.kts

KOTLINCopied
import org.jetbrains.kotlin.gradle.ExperimentalKotlinGradlePluginApi
import org.jetbrains.kotlin.gradle.dsl.JvmTarget

plugins {
    alias(libs.plugins.kotlinMultiplatform)
    alias(libs.plugins.androidLibrary)
    id("com.chaquo.python")
}

kotlin {
    androidTarget {
        @OptIn(ExperimentalKotlinGradlePluginApi::class)
        compilerOptions {
            jvmTarget.set(JvmTarget.JVM_11)
        }
    }

    listOf(
        iosX64(),
        iosArm64(),
        iosSimulatorArm64()
    ).forEach { iosTarget ->
        iosTarget.binaries.framework {
            baseName = "Shared"
            isStatic = true
        }
    }

    sourceSets {
        androidMain.dependencies {
        }
        iosMain.dependencies {
        }
        commonMain.dependencies {
            implementation(libs.multik.core)
            implementation(libs.multik.kotlin)
        }
    }
}

android {
    namespace = "com.assignment.shared"
    compileSdk = libs.versions.android.compileSdk.get().toInt()
    compileOptions {
        sourceCompatibility = JavaVersion.VERSION_11
        targetCompatibility = JavaVersion.VERSION_11
    }
    defaultConfig {
        minSdk = libs.versions.android.minSdk.get().toInt()
        ndk {
            // On Apple silicon, you can omit x86_64.
            abiFilters += listOf("arm64-v8a")
        }
    }
}
chaquopy {
    defaultConfig {
        buildPython ("/usr/local/bin/python3")
    }
    productFlavors { }
    sourceSets { }
}

I receive the following warning:

BASHCopied
Warning: Failed to compile to .pyc format: [/usr/local/bin/python3] does not appear to be a valid Python command. See https://chaquo.com/chaquopy/doc/current/android.html#android-bytecode

Although running python3 in the terminal works, importing Python in the shared module fails.

How can I adjust my Chaquopy configuration to successfully import Python (and access numpy) in my KMM application?

EDIT: I'm open to suggestions on alternative libraries or workarounds for processing .pkl files in a KMM project (iOS & Android), or converting the Python script into Kotlin/Multik.

Below is the Python script used for processing:

PYTHONCopied
# -*- coding: utf-8 -*-

import numpy as np
import pickle

def seg_stride(sig, dz, f_start, est_t):
    """! Estimate stride start and end for walking segment

    @param sig Moving standard deviation of heel
    @param dz Depth difference between heels
    @param f_start Start frame of walking segment
    @param est_t Estimated duration of one stride
    @return fp_lr Estimated frames for stride [start, end]
    """
    cur_len = len(sig)
    m_thres = np.mean(sig[sig < np.median(sig)]) # get the mean for values less than median
    [idx] = np.where(np.diff((sig < m_thres) * 1) > 0) # the search start point for each stride, where the heel z goes below threshold
    
    # Filter idx such that there is only one in each stride
    # upward zero-crossings to nearest time step
    [upcross] = np.where((dz[:-1] <= 0) & (dz[1:] > 0))
    # downward zero-crossings
    [downcross] = np.where((dz[:-1] >= 0) & (dz[1:] < 0))
    if len(upcross) >= len(downcross):
        cross = upcross
    else:
        cross = downcross
    cross = np.concatenate([[0], cross, [cur_len]])
    filtered_idx = []

    # Iterate through the intervals in downcross
    for i in range(len(cross) - 1):
        start = cross[i]
        end = cross[i + 1]
        # Find elements in idx that are in the interval (start, end)
        elements_in_interval = idx[(idx >= start) & (idx < end)]
        # If there is exactly one element, keep it; otherwise, keep the smallest element
        if len(elements_in_interval) >= 1:
            filtered_idx.append(elements_in_interval[0])
    
    # Convert filtered_idx back to a numpy array
    filtered_idx = np.array(filtered_idx)

    num_fp = len(filtered_idx) # no. of strides

    if num_fp > 1:
        fp = np.zeros((num_fp))
        # refine and record the corresponding frame
        for k in range(num_fp):
            cur_t = filtered_idx[k]
            cur_sig = sig[cur_t : min(cur_len, cur_t + round(0.25 * est_t))] # search space for one stride, start point to 1/4 of estimated stride or end of vid

            # locate the local minimum
            s_idx = np.argmin(cur_sig)
            t2 = min(cur_len, cur_t + s_idx)
            t1 = max(0, cur_t - round(0.2 * est_t)) # what is this for?
            t_idx = np.argmax(dz[t1 : t2 + 1]) # checking step to see if there is an earlier peak?
            fp[k] = f_start + t1 + t_idx

        # set fp_left: each row is [start frame, end frame] of a valid stride
        fp_lr = np.zeros((num_fp - 1, 2), dtype=int)
        for k in range(num_fp - 1):
            fp_lr[k, 0] = int(fp[k])
            fp_lr[k, 1] = int(fp[k + 1])

    else:
        fp_lr = np.empty((0, 2), dtype=int)

    return fp_lr


def dist2plane_new(points, ground_plane):
    """! Project 3D points to plane and compute shortest distance

    @param points Input 3D points
    @param ground_plane ground plane parameters (a,b,c,d)
    @return projected_points Projected points on ground plane
    @return distance Shortest distance between points and plane
    """

    a, b, c, d = ground_plane

    projected_points = []
    distance = []
    for point in points:
        s, u, v = point
        t = (d - a * s - b * u - c * v) / (a * a + b * b + c * c)

        # point closest to plane
        x = s + a * t
        y = u + b * t
        z = v + c * t

        dist = abs(a * s + b * u + c * z + d) / (np.sqrt(a * a + b * b + c * c))
        projected_points.append([x, y, z])
        distance.append(dist)

    return np.array(projected_points), np.array(distance)


def refine_stride_boundary(angle, foot_on, cur_t, c_margin, nb_frames):
    # set local segment
    t1 = max(0, cur_t - c_margin)
    t2 = min(nb_frames - 1, cur_t + c_margin)

    # refine the start of stride by finding the  largest decrease in heel angle
    idx = local_argmin_diff(angle, t1, t2)
    t = t1 + idx

    # further refine by ensuring foot is on ground at t
    if t not in foot_on:
        # if foot not on ground, adjust to next frame when foot is on ground
        next_foot_on = foot_on[(foot_on - t)>0]
        if len(next_foot_on)>0:
            t = next_foot_on[0]

    return t


def find_boundaries(fp, angle, angle_other, foot_on, est_t, nb_frames):
    c_margin = round(est_t / 10)
    t = [] 

    # refine stride boundaries (foot on)
    for k in range(len(fp)):
        # refine start and end of stride
        t_start = refine_stride_boundary(angle, foot_on, fp[k, 0], c_margin, nb_frames)
        t_end = refine_stride_boundary(angle, foot_on, fp[k, 1], c_margin, nb_frames)

        # find foot offs
        if t_end >= t_start+4: # if next foot on is at least 4 frames ahead
            # find foot off - min angle between start and end of stride
            t_fo = t_start + local_argmin(angle, t_start, t_end)
            # if current foot off is at least 2 frames ahead of start
            if t_fo >= t_start+2: 
                # find other foot off
                to_fo = t_start + local_argmin(angle_other, t_start, t_fo)
                t.append([t_start, to_fo, t_fo, t_end])

    return t


def combine_left_right(fp1, fp2):
    # foot ons
    t_strides = [t[0] for t in fp2] + [t[-1] for t in fp2]
    t_strides = sorted(set(t_strides))
    t_strides = np.array(t_strides)
    refined_fp = []
    for fp in fp1:
        other_foot_on = t_strides[(t_strides>fp[1]) & (t_strides<fp[2])]
        if len(other_foot_on) == 1:
            refined_fp.append([fp[0], fp[1], other_foot_on[0], fp[2], fp[3]])
    return np.array(refined_fp)


def params_est_lidar(pose_data, fp_walk_list, ground_plane):
    """! Estimate gait parameters based on 3D pose data by identifying corresponding timestamps

    @param pose_data 3D pose data (28 key points) for
    ('L_HEAD'    'R_HEAD'    'SGL'    'CV7'    'TV10'
    'L_SAE'     'R_SAE'    'MID_L_HE'    'MID_R_HE'    'MID_L_SP'
    'MID_R_SP' 'L_IAS'    'R_IAS'    'MID_IPS'    'L_FLE'
    'R_FLE'    'L_FME'    'R_FME'    'L_FAL'    'R_FAL'
    'L_TAM'    'R_TAM'    'L_FCC'     'R_FCC'    'L_FM2'    'R_FM2'    'L_FM5'    'R_FM5')
    @param fps Video frame rate
    @param fp_walk_list List of walking segments [[start1, end1, start2, end2], [s1,e1,s2,e4],...]
    @param ground_plane ground plane parameters (a,b,c,d)

    @return t_left Corresponding consecutive timestamps for left cycle
    @return t_right Corresponding consecutive timestamps for right cycle
    @return estimated stride lengths: stride_left, stride_right
    @return gait parameters output
    """

    # set 3D points
    left_heel = pose_data[:, 22, :]  # L_FCC
    right_heel = pose_data[:, 23, :]  # R_FCC
    left_toe = pose_data[:, 24, :]  # L_FM2 (idx: 25)
    right_toe = pose_data[:, 25, :]  # R_FM2 (idx: 26)

    # get number of frames
    nb_frames = pose_data.shape[0]

    # finding boundaries of each stride
    # depth difference
    dz = right_heel[:, 2] - left_heel[:, 2]
    # upward zero-crossings to nearest time step
    [upcross] = np.where((dz[:-1] <= 0) & (dz[1:] > 0))
    # downward zero-crossings
    [downcross] = np.where((dz[:-1] >= 0) & (dz[1:] < 0))

    # estimate the duration of one stride
    diff_up = np.diff(upcross).tolist()
    diff_down = np.diff(downcross).tolist()
    est_t = np.median(diff_up + diff_down)
    print('The estimated duration of one stride is %d frames' % est_t)

    # use the 2D-based phase segmentation results to set walking segments (change on 02/07/2023)
    # take input list and run in loop for walk-turn-walk (22/01/2024)
    fp_left = []
    fp_right = []
    print("number of complete round of walk-forward, walk-backward: ", len(fp_walk_list))
    for fp_walk in fp_walk_list:
        f_start1 = fp_walk[0]
        f_end1 = fp_walk[1]
        f_start2 = fp_walk[2]
        f_end2 = fp_walk[3]

        # for 1st walking segment (walk towards camera)
        # for left foot
        sig1 = left_heel[f_start1 : f_end1 + 1, 2]  # relative-depth (05/Oct/2023)
        dz1 = dz[f_start1 : f_end1 + 1]
        fp1_left = seg_stride(sig1, dz1, f_start1, est_t)

        # for right foot
        sig2 = right_heel[f_start1 : f_end1 + 1, 2]  # relative-depth (05/Oct/2023)
        dz2 = -dz1
        fp1_right = seg_stride(sig2, dz2, f_start1, est_t)

        # for 2nd walking segment (walk away from camera)
        # for left foot
        sig1 = -left_heel[f_start2 : f_end2 + 1, 2]  # add negative
        dz1 = -dz[f_start2 : f_end2 + 1]
        fp2_left = seg_stride(sig1, dz1, f_start2, est_t)

        # for right foot
        sig2 = -right_heel[f_start2 : f_end2 + 1, 2]  # relative-depth (05/Oct/2023)
        dz2 = -dz1
        fp2_right = seg_stride(sig2, dz2, f_start2, est_t)

        fp_left.append(fp1_left)
        fp_left.append(fp2_left)
        fp_right.append(fp1_right)
        fp_right.append(fp2_right)
    #print('Finished for loop')
    fp_left = np.vstack(fp_left)
    fp_right = np.vstack(fp_right)

    # Further processing to identify single support, double support, and refine stride boundaries
    #print('Starting Step3')
    # Step 3: Calculate distance to plane and angles between foot and floor

    # angle between left foot and floor plane
    [pp1, dz1] = dist2plane_new(left_heel, ground_plane)
    [pp2, dz2] = dist2plane_new(left_toe, ground_plane)
    perpendicular = dz2 - dz1 # height difference between left heel and toe relative to ground plane
    base = np.sqrt(np.sum(np.power(pp2 - pp1, 2), 1))  # distance between left heel and left toe
    left_angle = np.arctan2(perpendicular, base)
    left_angle = left_angle * 180 / np.pi  # convert to degree
    
    # angle between right foot and floor plane
    [pp1, dz1] = dist2plane_new(right_heel, ground_plane)
    [pp2, dz2] = dist2plane_new(right_toe, ground_plane)
    perpendicular = dz2 - dz1
    base = np.sqrt(np.sum(np.power(pp2 - pp1, 2), 1))
    right_angle = np.arctan2(perpendicular, base)
    right_angle = right_angle * 180 / np.pi  # convert to degree

    # distance between ankles and floor
    [ppl, dzl] = dist2plane_new(left_heel, ground_plane)
    [ppr, dzr] = dist2plane_new(right_heel, ground_plane)
    # compute when each foot is on ground
    left_foot_on = np.where(dzl < dzr)[0]
    right_foot_on = np.where(dzr < dzl)[0]

    #print('Starting Step4')
    # Step 4: refine boundaries of each stride and further divide into
    # double support (DS1), single support, and double support (DS2)

    # refine stride boundaries (foot on) and find foot offs
    t_left = find_boundaries(fp_left, left_angle, right_angle, left_foot_on, est_t, nb_frames)
    t_right = find_boundaries(fp_right, right_angle, left_angle, right_foot_on, est_t, nb_frames)

    # combine left and right times
    t_left = combine_left_right(t_left, t_right)
    t_right = combine_left_right(t_right, t_left)

    return t_left, t_right  # , stride_left, stride_right, output

def local_argmin(sig, start, end):
    """! Find local minimum index

    @param sig Signal
    @return Local minimum index
    """
    if end <= start:
        idx = 0
    else:
        idx = np.argmin(sig[start : end + 1])
    
    return idx


def local_argmin_diff(sig, start, end):
    """! Find local minimum index of difference

    @param sig Signal
    @return Local minimum index
    """
    if end <= start:
        idx = 0
    else:
        idx = np.argmin(np.diff(sig[start : end + 1]))
    
    return idx

with open('input_output/in3.pkl','rb') as file:
    data = pickle.load(file) 

print(f'{data}')
pose_data_3d_norm = data['pose_data_3d_norm']
fp_walk = data['fp_walk']
ground_plane = data['ground_plane']

[t_left, t_right] = params_est_lidar(pose_data_3d_norm, fp_walk, ground_plane)