<filename>trackAnalyzer/Features.py
#!/usr/bin/env python

import os, pdb
import numpy as np
from scipy import interpolate
from node_graph import Graph
import math
import uuid

# Superclass for Poles and tracks that stores positional and intensity information
class Feature:
    def __init__(self, time, position, intensity, strain='xxx'):
        self.time = np.array( time )
        self.position = np.array( position )
        self.intensity = np.array( intensity )
        self.id = uuid.uuid1()
        self.strain = strain

# Class for a Pole
class Pole(Feature):
    def __init__(self, time, position, intensity, strain='xxx'):
        Feature.__init__(self, time, position, intensity, strain)

        # Define an interpolation/extrapolation function
        self.ifunc = interpolate.interp1d(self.time, self.position, kind='linear', fill_value='extrapolate')

    def Print(self):
        print('Pole :')
        print(' ID : {}'.format(self.id))
        print(' Time : {}'.format( self.time))
        print(' Position : {}'.format( self.position))
        print(' Intensity : {}'.format( self.intensity))
        print('--------------------------------- ')

# Class for a Track: additionally stores associated poles and track direction
class Track(Feature):
    def __init__(self, time, position, intensity, poles, direction, line_type, strain='xxx'):
        Feature.__init__(self, time, position, intensity, strain)
        self.poles = poles

        self.direction = direction
        self.line_type = line_type
        self.polePosition = []
        self.data = {
                'pos_pole' : np.zeros( (2, np.size(self.position) ) ),
                'pos_track_rel' : [],
                'velocity' : {
                    'P' : [],
                    'AP' : [],
                    },
                'runlength' : {
                    'P' : [],
                    'AP' : [],
                    },
                'lifetime' : {
                    'P' : [],
                    'AP' : [],
                    },
                'lifetime_total' : [],
                'velocity_mean' : [],
                'switch_count' : [],
                'switch_total' : [],
                }
        self.bad = 0

        # # If track has a direction, then set its line_type to Line
        # if self.line_type == 'Curve' and (self.direction == 'Poleward' or self.direction == 'Antipoleward'):
            # pdb.set_trace()
            # self.line_type = 'Line'
        if not self.poles:
            pdb.set_trace()
            print('1')

    def Analyze(self):
    # Run useful analysis methods

        self.CalcPositionTrackRelativeToPole()

        # Split the track and save analysis 
        tracks_mini, switch_count = self.SplitTrack()

        for track in tracks_mini:

            # Ensure mini track is a line
            # if track.line_type != 'Line':
                # pdb.set_trace()
                # raise Exception('mini track must be a line')

            # Calculate and append data of the mini track
            if track.direction == 'Poleward':

                # Velocity
                self.data['velocity']['P'] += [track.CalcVelocityLinear()]
                # Run length
                self.data['runlength']['P'] += [track.CalcRunLength()]
                # Lifetime
                self.data['lifetime']['P'] += [track.CalcLifetime()]

            elif track.direction == 'Antipoleward':

                # Velocity
                self.data['velocity']['AP'] += [track.CalcVelocityLinear()]
                # Run length
                self.data['runlength']['AP'] += [track.CalcRunLength()]
                # Lifetime
                self.data['lifetime']['AP'] += [track.CalcLifetime()]

            else:
                pdb.set_trace()
                raise ValueError('line direction is neither poleward nor antipoleward')

        # Combine data from the split tracks
        self.data['lifetime_total'] = self.CalcLifetime()
        self.data['velocity_mean'] = self.CalcVelocityMean()
        self.data['switch_count'] = switch_count
        self.data['switch_total'] = switch_count['P'] + switch_count['AP'] 

    def CheckViability(self):
        # Check if the track's time points are always increasing

        self.bad = 0
        # Check track time is always increasing
        if np.any( np.diff( self.time) <= 0 ):
            self.bad = 1 

        return self.bad

    def CheckLinearRunLength( self, min_rl = 0.01):
        # Check run length is above a min threshold

        self.bad = 0
        if self.line_type == 'Line' and self.CalcRunLength() < min_rl:
            self.bad = 1

        return self.bad

    def CheckLinearLifetime( self, min_lt = 1):
        # Check lifetime is above a min threshold

        self.bad = 0
        if self.line_type == 'Line' and self.CalcLifetime() < min_lt:
            self.bad = 1

        return self.bad


    def CalcPositionPoleCurrent(self):
    # Get pole position at the current time (i.e at the times of the track) by using the interpolation/extrapolation function of the pole

        for idx, pole in enumerate( self.poles) :
            pos_pole = np.array(pole.ifunc( self.time) )
            self.data['pos_pole'][idx,:] = pos_pole
        return self.data['pos_pole'] 

    def CalcPositionTrackRelativeToPole(self):
    # Calculate track position relative to the pole 

        if not self.data['pos_pole'].any():
            pos_pole = self.CalcPositionPoleCurrent()
        else:
            pos_pole = self.data['pos_pole']

        # If bipolar spindle
        if len( self.poles) == 2:
            print('Not sure how to calculate relative position in a bipolar spindle')

        # If monopolar spindle
        else:
            pos_track_rel = np.array( self.position - pos_pole )
            pos_track_rel = pos_track_rel[0,:]
            self.data['pos_track_rel'] = pos_track_rel

        return pos_track_rel

    def CalcVelocityLinear(self):
    # Calculate the velocity of this linear track
        
        if self.direction == 'Ambiguous':
            raise Exception('Track.CalcVelocityLinear() is only defined for tracks with a single direction')

        # Calc relative positions if not done already
        if len( self.data['pos_track_rel']) == 0 or not self.data['pos_track_rel'].any():
            pos_track_rel = self.CalcPositionTrackRelativeToPole()
        else:
            pos_track_rel = self.data['pos_track_rel']

        # if np.isin( np.diff(self.time), 0).any():
            # print('1')

        vel = np.average( np.absolute( np.divide( np.diff( pos_track_rel) , np.diff( self.time) ) ), weights = np.diff(self.time) )
        if np.size( vel) > 1:
            pdb.set_trace()
            print('1')
        return vel
    
    def CalcRunLength(self):
    # Calculate the run length of this track
        
        # Calc relative positions if not done already
        if len( self.data['pos_track_rel']) == 0 or not self.data['pos_track_rel'].any():
            pos_track_rel = self.CalcPositionTrackRelativeToPole()
        else:
            pos_track_rel = self.data['pos_track_rel']

        run_length = np.absolute( pos_track_rel[-1] - pos_track_rel[0] )
        self.data['run_length'] = run_length
        if np.size( run_length) > 1:
            pdb.set_trace()
            print('1')
        return run_length

    def CalcLifetime(self):
    # Calculate the lifetime of this track

        lifetime = self.time[-1] - self.time[0]
        self.data['lifetime'] = lifetime
        return lifetime

    def CalcVelocityMean(self):
    # Calculate the mean velocity of this track

        if self.line_type == 'Curve' and not self.data['velocity']:
            # Split the track 
            tracks_mini = self.SplitTrack()
            for track in tracks_mini:
                if track.direc == 'Poleward':
                    self.data['velocity']['P'] += [track.CalcVelocityLinear()]
                elif self.direc == 'Antipoleward':
                    self.data['velocity']['AP'] += [track.CalcVelocityLinear()]
        
        vel_mu = np.mean( np.concatenate( (self.data['velocity']['P'], self.data['velocity']['AP']) ) )
        return vel_mu

    def CalcSwitchingCount(self):
    # Calculate the mean velocity of this track

        if self.line_type == 'Curve' and not self.data['velocity']:
            # Split the track 
            tracks_mini = self.SplitTrack()
            for track in tracks_mini:
                if track.direc == 'Poleward':
                    self.data['velocity']['P'] += [track.CalcVelocityLinear()]
                elif self.direc == 'Antipoleward':
                    self.data['velocity']['AP'] += [track.CalcVelocityLinear()]
        
        vel_mu = np.mean( np.concat( self.data['velocity']['P'], self.data['velocity']['AP']) )
        return vel_mu

    def CalcIntensityMean(self):
    # Calculate the mean intensity of this track
        self.data['intensity_mean'] = np.mean( self.intensity) 
        
    def SplitTrack(self):
    # Split curved track into multiple mini unidirectional tracks

        switch_count = {
                'P' : 0,
                'AP': 0,
                }

        if self.line_type == 'Line' or self.direction != 'Ambiguous':
            return [self], switch_count 

        index_switch = []
   
        # Loop over track position differences and find indices of direction switches
        dx = np.diff( self.position - self.poles[0].ifunc( self.time) )
        for idx,pos in enumerate( dx[0:-1]):

            # if sign of derivative is different for next element, then store index as a switching index 
            if np.sign( pos ) != np.sign( dx[idx+1]):
                index_switch += [idx+1]

        # Initialize mini tracks
        mini_tracks = []
        index_tracks = [0] + index_switch + [len(self.position)-1] 

        for idx,jt in enumerate( index_tracks[0:-1]):
            
            time = self.time[ jt : index_tracks[idx+1]+1 ]
            pos = self.position[ jt : index_tracks[idx+1]+1 ]
            intensity = self.intensity[ jt : index_tracks[idx+1]+1 ]

            # If Monopolar Spindle
            if len(self.poles) == 1:

                # Determine if track goes towards or away from pole 
                if np.mean( np.diff( np.absolute( pos - self.poles[0].ifunc( time) ) ) ) > 0:
                    direc = 'Antipoleward'
                else:
                    direc = 'Poleward'

            # If Bipolar Spindle
            else: 
                pdb.set_trace()
                # Find spindle midpoint
                midpt = np.array( (self.poles[0].ifunc( time) + self.poles[1].ifunc( time))/2 )

                # Determine if track goes towards spindle mid point or away from spindle midpoint
                if np.mean( np.diff( pos - midpt) ) < 0:
                    direc = 'Antipoleward'
                else:
                    direc = 'Poleward'

            if jt != index_tracks[0] and jt != index_tracks[-1]:
                if direc == 'Poleward':
                    switch_count['P'] += 1
                elif direc == 'Antipoleward':
                    switch_count['AP'] += 1

            mini_tracks += [Track(time, pos, intensity, self.poles, direc, 'Line', strain=self.strain) ]
        return mini_tracks, switch_count

    def Print(self):
        print('Feature :')
        print(' ID : {}'.format(self.id))
        print(' Direction : {}'.format( self.direction))
        print(' Line type : {}'.format( self.line_type))
        print(' Time : {}'.format( self.time))
        print(' Position : {}'.format( self.position))
        print(' Intensity : {}'.format( self.intensity))
        print('--------------------------------- ')



<filename>trackAnalyzer/Features.py
#!/usr/bin/env python

import os, pdb
import numpy as np
from scipy import interpolate
from node_graph import Graph
import math
import uuid

# Superclass for Poles and tracks that stores positional and intensity information
class Feature:
    def __init__(self, time, position, intensity, strain='xxx'):
        self.time = np.array( time )
        self.position = np.array( position )
        self.intensity = np.array( intensity )
        self.id = uuid.uuid1()
        self.strain = strain

# Class for a Pole
class Pole(Feature):
    def __init__(self, time, position, intensity, strain='xxx'):
        Feature.__init__(self, time, position, intensity, strain)

        # Define an interpolation/extrapolation function
        self.ifunc = interpolate.interp1d(self.time, self.position, kind='linear', fill_value='extrapolate')

    def Print(self):
        print('Pole :')
        print(' ID : {}'.format(self.id))
        print(' Time : {}'.format( self.time))
        print(' Position : {}'.format( self.position))
        print(' Intensity : {}'.format( self.intensity))
        print('--------------------------------- ')

# Class for a Track: additionally stores associated poles and track direction
class Track(Feature):
    def __init__(self, time, position, intensity, poles, direction, line_type, strain='xxx'):
        Feature.__init__(self, time, position, intensity, strain)
        self.poles = poles

        self.direction = direction
        self.line_type = line_type
        self.polePosition = []
        self.data = {
                'pos_pole' : np.zeros( (2, np.size(self.position) ) ),
                'pos_track_rel' : [],
                'velocity' : {
                    'P' : [],
                    'AP' : [],
                    },
                'runlength' : {
                    'P' : [],
                    'AP' : [],
                    },
                'lifetime' : {
                    'P' : [],
                    'AP' : [],
                    },
                'lifetime_total' : [],
                'velocity_mean' : [],
                'switch_count' : [],
                'switch_total' : [],
                }
        self.bad = 0

        # # If track has a direction, then set its line_type to Line
        # if self.line_type == 'Curve' and (self.direction == 'Poleward' or self.direction == 'Antipoleward'):
            # pdb.set_trace()
            # self.line_type = 'Line'
        if not self.poles:
            pdb.set_trace()
            print('1')

    def Analyze(self):
    # Run useful analysis methods

        self.CalcPositionTrackRelativeToPole()

        # Split the track and save analysis 
        tracks_mini, switch_count = self.SplitTrack()

        for track in tracks_mini:

            # Ensure mini track is a line
            # if track.line_type != 'Line':
                # pdb.set_trace()
                # raise Exception('mini track must be a line')

            # Calculate and append data of the mini track
            if track.direction == 'Poleward':

                # Velocity
                self.data['velocity']['P'] += [track.CalcVelocityLinear()]
                # Run length
                self.data['runlength']['P'] += [track.CalcRunLength()]
                # Lifetime
                self.data['lifetime']['P'] += [track.CalcLifetime()]

            elif track.direction == 'Antipoleward':

                # Velocity
                self.data['velocity']['AP'] += [track.CalcVelocityLinear()]
                # Run length
                self.data['runlength']['AP'] += [track.CalcRunLength()]
                # Lifetime
                self.data['lifetime']['AP'] += [track.CalcLifetime()]

            else:
                pdb.set_trace()
                raise ValueError('line direction is neither poleward nor antipoleward')

        # Combine data from the split tracks
        self.data['lifetime_total'] = self.CalcLifetime()
        self.data['velocity_mean'] = self.CalcVelocityMean()
        self.data['switch_count'] = switch_count
        self.data['switch_total'] = switch_count['P'] + switch_count['AP'] 

    def CheckViability(self):
        # Check if the track's time points are always increasing

        self.bad = 0
        # Check track time is always increasing
        if np.any( np.diff( self.time) <= 0 ):
            self.bad = 1 

        return self.bad

    def CheckLinearRunLength( self, min_rl = 0.01):
        # Check run length is above a min threshold

        self.bad = 0
        if self.line_type == 'Line' and self.CalcRunLength() < min_rl:
            self.bad = 1

        return self.bad

    def CheckLinearLifetime( self, min_lt = 1):
        # Check lifetime is above a min threshold

        self.bad = 0
        if self.line_type == 'Line' and self.CalcLifetime() < min_lt:
            self.bad = 1

        return self.bad


    def CalcPositionPoleCurrent(self):
    # Get pole position at the current time (i.e at the times of the track) by using the interpolation/extrapolation function of the pole

        for idx, pole in enumerate( self.poles) :
            pos_pole = np.array(pole.ifunc( self.time) )
            self.data['pos_pole'][idx,:] = pos_pole
        return self.data['pos_pole'] 

    def CalcPositionTrackRelativeToPole(self):
    # Calculate track position relative to the pole 

        if not self.data['pos_pole'].any():
            pos_pole = self.CalcPositionPoleCurrent()
        else:
            pos_pole = self.data['pos_pole']

        # If bipolar spindle
        if len( self.poles) == 2:
            print('Not sure how to calculate relative position in a bipolar spindle')

        # If monopolar spindle
        else:
            pos_track_rel = np.array( self.position - pos_pole )
            pos_track_rel = pos_track_rel[0,:]
            self.data['pos_track_rel'] = pos_track_rel

        return pos_track_rel

    def CalcVelocityLinear(self):
    # Calculate the velocity of this linear track
        
        if self.direction == 'Ambiguous':
            raise Exception('Track.CalcVelocityLinear() is only defined for tracks with a single direction')

        # Calc relative positions if not done already
        if len( self.data['pos_track_rel']) == 0 or not self.data['pos_track_rel'].any():
            pos_track_rel = self.CalcPositionTrackRelativeToPole()
        else:
            pos_track_rel = self.data['pos_track_rel']

        # if np.isin( np.diff(self.time), 0).any():
            # print('1')

        vel = np.average( np.absolute( np.divide( np.diff( pos_track_rel) , np.diff( self.time) ) ), weights = np.diff(self.time) )
        if np.size( vel) > 1:
            pdb.set_trace()
            print('1')
        return vel
    
    def CalcRunLength(self):
    # Calculate the run length of this track
        
        # Calc relative positions if not done already
        if len( self.data['pos_track_rel']) == 0 or not self.data['pos_track_rel'].any():
            pos_track_rel = self.CalcPositionTrackRelativeToPole()
        else:
            pos_track_rel = self.data['pos_track_rel']

        run_length = np.absolute( pos_track_rel[-1] - pos_track_rel[0] )
        self.data['run_length'] = run_length
        if np.size( run_length) > 1:
            pdb.set_trace()
            print('1')
        return run_length

    def CalcLifetime(self):
    # Calculate the lifetime of this track

        lifetime = self.time[-1] - self.time[0]
        self.data['lifetime'] = lifetime
        return lifetime

    def CalcVelocityMean(self):
    # Calculate the mean velocity of this track

        if self.line_type == 'Curve' and not self.data['velocity']:
            # Split the track 
            tracks_mini = self.SplitTrack()
            for track in tracks_mini:
                if track.direc == 'Poleward':
                    self.data['velocity']['P'] += [track.CalcVelocityLinear()]
                elif self.direc == 'Antipoleward':
                    self.data['velocity']['AP'] += [track.CalcVelocityLinear()]
        
        vel_mu = np.mean( np.concatenate( (self.data['velocity']['P'], self.data['velocity']['AP']) ) )
        return vel_mu

    def CalcSwitchingCount(self):
    # Calculate the mean velocity of this track

        if self.line_type == 'Curve' and not self.data['velocity']:
            # Split the track 
            tracks_mini = self.SplitTrack()
            for track in tracks_mini:
                if track.direc == 'Poleward':
                    self.data['velocity']['P'] += [track.CalcVelocityLinear()]
                elif self.direc == 'Antipoleward':
                    self.data['velocity']['AP'] += [track.CalcVelocityLinear()]
        
        vel_mu = np.mean( np.concat( self.data['velocity']['P'], self.data['velocity']['AP']) )
        return vel_mu

    def CalcIntensityMean(self):
    # Calculate the mean intensity of this track
        self.data['intensity_mean'] = np.mean( self.intensity) 
        
    def SplitTrack(self):
    # Split curved track into multiple mini unidirectional tracks

        switch_count = {
                'P' : 0,
                'AP': 0,
                }

        if self.line_type == 'Line' or self.direction != 'Ambiguous':
            return [self], switch_count 

        index_switch = []
   
        # Loop over track position differences and find indices of direction switches
        dx = np.diff( self.position - self.poles[0].ifunc( self.time) )
        for idx,pos in enumerate( dx[0:-1]):

            # if sign of derivative is different for next element, then store index as a switching index 
            if np.sign( pos ) != np.sign( dx[idx+1]):
                index_switch += [idx+1]

        # Initialize mini tracks
        mini_tracks = []
        index_tracks = [0] + index_switch + [len(self.position)-1] 

        for idx,jt in enumerate( index_tracks[0:-1]):
            
            time = self.time[ jt : index_tracks[idx+1]+1 ]
            pos = self.position[ jt : index_tracks[idx+1]+1 ]
            intensity = self.intensity[ jt : index_tracks[idx+1]+1 ]

            # If Monopolar Spindle
            if len(self.poles) == 1:

                # Determine if track goes towards or away from pole 
                if np.mean( np.diff( np.absolute( pos - self.poles[0].ifunc( time) ) ) ) > 0:
                    direc = 'Antipoleward'
                else:
                    direc = 'Poleward'

            # If Bipolar Spindle
            else: 
                pdb.set_trace()
                # Find spindle midpoint
                midpt = np.array( (self.poles[0].ifunc( time) + self.poles[1].ifunc( time))/2 )

                # Determine if track goes towards spindle mid point or away from spindle midpoint
                if np.mean( np.diff( pos - midpt) ) < 0:
                    direc = 'Antipoleward'
                else:
                    direc = 'Poleward'

            if jt != index_tracks[0] and jt != index_tracks[-1]:
                if direc == 'Poleward':
                    switch_count['P'] += 1
                elif direc == 'Antipoleward':
                    switch_count['AP'] += 1

            mini_tracks += [Track(time, pos, intensity, self.poles, direc, 'Line', strain=self.strain) ]
        return mini_tracks, switch_count

    def Print(self):
        print('Feature :')
        print(' ID : {}'.format(self.id))
        print(' Direction : {}'.format( self.direction))
        print(' Line type : {}'.format( self.line_type))
        print(' Time : {}'.format( self.time))
        print(' Position : {}'.format( self.position))
        print(' Intensity : {}'.format( self.intensity))
        print('--------------------------------- ')



#!/usr/bin/env python # Superclass for Poles and tracks that stores positional and intensity information # Class for a Pole # Define an interpolation/extrapolation function # Class for a Track: additionally stores associated poles and track direction # # If track has a direction, then set its line_type to Line # if self.line_type == 'Curve' and (self.direction == 'Poleward' or self.direction == 'Antipoleward'): # pdb.set_trace() # self.line_type = 'Line' # Run useful analysis methods # Split the track and save analysis # Ensure mini track is a line # if track.line_type != 'Line': # pdb.set_trace() # raise Exception('mini track must be a line') # Calculate and append data of the mini track # Velocity # Run length # Lifetime # Velocity # Run length # Lifetime # Combine data from the split tracks # Check if the track's time points are always increasing # Check track time is always increasing # Check run length is above a min threshold # Check lifetime is above a min threshold # Get pole position at the current time (i.e at the times of the track) by using the interpolation/extrapolation function of the pole # Calculate track position relative to the pole # If bipolar spindle # If monopolar spindle # Calculate the velocity of this linear track # Calc relative positions if not done already # if np.isin( np.diff(self.time), 0).any(): # print('1') # Calculate the run length of this track # Calc relative positions if not done already # Calculate the lifetime of this track # Calculate the mean velocity of this track # Split the track # Calculate the mean velocity of this track # Split the track # Calculate the mean intensity of this track # Split curved track into multiple mini unidirectional tracks # Loop over track position differences and find indices of direction switches # if sign of derivative is different for next element, then store index as a switching index # Initialize mini tracks # If Monopolar Spindle # Determine if track goes towards or away from pole # If Bipolar Spindle # Find spindle midpoint # Determine if track goes towards spindle mid point or away from spindle midpoint