<reponame>Cocopyth/MscThesis<filename>amftrack/pipeline/functions/post_processing/time_hypha.py<gh_stars>1-10
from amftrack.pipeline.functions.image_processing.hyphae_id_surf import get_pixel_growth_and_new_children
import numpy as np
import networkx as nx
from amftrack.pipeline.functions.post_processing.util import get_length_um_edge, is_in_study_zone,get_length_um
from scipy import sparse

def get_time_since_start(hypha, t, tp1, args):
    exp = hypha.experiment
    seconds = (exp.dates[t]-exp.dates[hypha.ts[0]]).total_seconds()
    return("time_since_emergence",seconds/3600)

def get_time_since_begin_exp(hypha, t, tp1, args):
    exp = hypha.experiment
    seconds = (exp.dates[t]-exp.dates[0]).total_seconds()
    return("time_since_begin_exp",seconds/3600)

def get_timedelta(hypha,t,tp1,args):
    exp = hypha.experiment
    seconds = (exp.dates[tp1]-exp.dates[t]).total_seconds()
    return("timedelta",seconds/3600)


def get_speed(hypha,t,tp1,args):
    try:
        pixels,nodes = get_pixel_growth_and_new_children(hypha,t,tp1)
        speed = np.sum([get_length_um(seg) for seg in pixels])/get_timedelta(hypha,t,tp1,None)[1]
        return('speed',speed)
    except nx.exception.NetworkXNoPath:
        print('not_connected',hypha.end.label,hypha.get_root(tp1).label)
        return('speed',None)

def get_timestep(hypha,t,tp1,args):
    return("timestep",t)

def get_timestep_init(hypha,t,tp1,args):
    return("timestep_init",hypha.ts[0])

def get_time_init(hypha,t,tp1,args):
    return("time_init",get_timedelta(hypha,0,hypha.ts[0],None)[1])

def get_degree(hypha,t,tp1,args):
    return("degree",hypha.end.degree(t))

def get_width_tip_edge(hypha,t,tp1,args):
    try:
        edges = hypha.get_nodes_within(t)[1]
        if len(edges)>0:
            edge = edges[-1]
            width = edge.width(t)
            return('width_tip_edge',width)
        else:
            return('width_tip_edge',None)
    except nx.exception.NetworkXNoPath:
        return('width_tip_edge',None)
    
def get_width_root_edge(hypha,t,tp1,args):
    try:
        edges = hypha.get_nodes_within(t)[1]
        if len(edges)>0:
            edge = edges[0]
            width = edge.width(t)
            return('width_tip_edge',width)
        else:
            return('width_tip_edge',None)
    except nx.exception.NetworkXNoPath:
        return('width_root_edge',None)
    
def get_width_average(hypha,t,tp1,args):
    try:
        edges = hypha.get_nodes_within(t)[1]
        widths = np.array([edge.width(t) for edge in edges])
        lengths = np.array([get_length_um_edge(edge,t) for edge in edges])
        av_width = np.sum(widths*lengths)/np.sum(lengths)
        return('av_width',av_width)
    except nx.exception.NetworkXNoPath:
        return('av_width',None)
    
def has_reached_final_pos(hypha,t,tp1,args):
    tf = hypha.ts[-1]
    pos_end = hypha.end.pos(tf)
    thresh = 40
    return('has_reached_final_pos',str(np.linalg.norm(hypha.end.pos(t)-pos_end)<thresh))

def get_distance_final_pos(hypha,t,tp1,args):
    tf = hypha.ts[-1]
    pos_end = hypha.end.pos(tf)
    return('distance_final_pos',np.linalg.norm(hypha.end.pos(t)-pos_end))


def local_density(hypha,t,tp1,args):
    window = args[0]
    exp = hypha.experiment
    pos = hypha.end.pos(t)
    x,y = pos[0],pos[1]
    skeletons = [sparse.csr_matrix(skel) for skel in exp.skeletons]
    skeleton=skeletons[t][x-window:x+window,y-window:y+window]
    density = skeleton.count_nonzero()/(window**2*1.725)*10**6
    return(f'density_window{window}',density)

def in_ROI(hypha,t,tp1,args):
    return('in_ROI', str(np.all(is_in_study_zone(hypha.end,t,1000,150))))

from amftrack.pipeline.functions.image_processing.hyphae_id_surf import get_pixel_growth_and_new_children
import numpy as np
import networkx as nx
from amftrack.pipeline.functions.post_processing.util import get_length_um_edge, is_in_study_zone,get_length_um
from scipy import sparse

def get_time_since_start(hypha, t, tp1, args):
    exp = hypha.experiment
    seconds = (exp.dates[t]-exp.dates[hypha.ts[0]]).total_seconds()
    return("time_since_emergence",seconds/3600)

def get_time_since_begin_exp(hypha, t, tp1, args):
    exp = hypha.experiment
    seconds = (exp.dates[t]-exp.dates[0]).total_seconds()
    return("time_since_begin_exp",seconds/3600)

def get_timedelta(hypha,t,tp1,args):
    exp = hypha.experiment
    seconds = (exp.dates[tp1]-exp.dates[t]).total_seconds()
    return("timedelta",seconds/3600)


def get_speed(hypha,t,tp1,args):
    try:
        pixels,nodes = get_pixel_growth_and_new_children(hypha,t,tp1)
        speed = np.sum([get_length_um(seg) for seg in pixels])/get_timedelta(hypha,t,tp1,None)[1]
        return('speed',speed)
    except nx.exception.NetworkXNoPath:
        print('not_connected',hypha.end.label,hypha.get_root(tp1).label)
        return('speed',None)

def get_timestep(hypha,t,tp1,args):
    return("timestep",t)

def get_timestep_init(hypha,t,tp1,args):
    return("timestep_init",hypha.ts[0])

def get_time_init(hypha,t,tp1,args):
    return("time_init",get_timedelta(hypha,0,hypha.ts[0],None)[1])

def get_degree(hypha,t,tp1,args):
    return("degree",hypha.end.degree(t))

def get_width_tip_edge(hypha,t,tp1,args):
    try:
        edges = hypha.get_nodes_within(t)[1]
        if len(edges)>0:
            edge = edges[-1]
            width = edge.width(t)
            return('width_tip_edge',width)
        else:
            return('width_tip_edge',None)
    except nx.exception.NetworkXNoPath:
        return('width_tip_edge',None)
    
def get_width_root_edge(hypha,t,tp1,args):
    try:
        edges = hypha.get_nodes_within(t)[1]
        if len(edges)>0:
            edge = edges[0]
            width = edge.width(t)
            return('width_tip_edge',width)
        else:
            return('width_tip_edge',None)
    except nx.exception.NetworkXNoPath:
        return('width_root_edge',None)
    
def get_width_average(hypha,t,tp1,args):
    try:
        edges = hypha.get_nodes_within(t)[1]
        widths = np.array([edge.width(t) for edge in edges])
        lengths = np.array([get_length_um_edge(edge,t) for edge in edges])
        av_width = np.sum(widths*lengths)/np.sum(lengths)
        return('av_width',av_width)
    except nx.exception.NetworkXNoPath:
        return('av_width',None)
    
def has_reached_final_pos(hypha,t,tp1,args):
    tf = hypha.ts[-1]
    pos_end = hypha.end.pos(tf)
    thresh = 40
    return('has_reached_final_pos',str(np.linalg.norm(hypha.end.pos(t)-pos_end)<thresh))

def get_distance_final_pos(hypha,t,tp1,args):
    tf = hypha.ts[-1]
    pos_end = hypha.end.pos(tf)
    return('distance_final_pos',np.linalg.norm(hypha.end.pos(t)-pos_end))


def local_density(hypha,t,tp1,args):
    window = args[0]
    exp = hypha.experiment
    pos = hypha.end.pos(t)
    x,y = pos[0],pos[1]
    skeletons = [sparse.csr_matrix(skel) for skel in exp.skeletons]
    skeleton=skeletons[t][x-window:x+window,y-window:y+window]
    density = skeleton.count_nonzero()/(window**2*1.725)*10**6
    return(f'density_window{window}',density)

def in_ROI(hypha,t,tp1,args):
    return('in_ROI', str(np.all(is_in_study_zone(hypha.end,t,1000,150))))