"""Quantify metabolic interactions between taxa."""from..taxonomyimporttaxon_idfrom..workflowsimportGrowthResults,workflowimportpandasaspdfromtypingimportList,Union
[docs]def_metabolite_interaction(fluxes:pd.DataFrame,taxon:str,partner:str)->pd.DataFrame:"""Checks if and how taxa interact."""tol=fluxes.tolerance.max()f=fluxes[(fluxes.flux.abs()*fluxes.abundance)>tol]if(f.shape[0]<2)or(f.direction=="export").all():returnNoneif(f.direction=="import").sum()==2:int_type="co-consumed"elif(f.loc[f.taxon==taxon,"direction"]=="export").all():int_type="provided"else:int_type="received"returnpd.DataFrame({"focal":taxon,"partner":partner,"class":int_type,"flux":(f.flux.abs()*f.abundance).min(),},index=[0],)
[docs]defsample_interactions(fluxes:pd.DataFrame,sample_id:str,taxon:str)->pd.DataFrame:"""Quantify interactions in a single sammple. Arguments --------- fluxes : pandas.DataFrame A table of exchange fluxes. sample_id : str The sample id to use. taxon : str The focal taxon to use. Returns ------- pandas.DataFrame The mapped interactions between the focal taxon and all other taxa. """ex=fluxes[fluxes.sample_id==sample_id]partners=pd.Series(ex.taxon.unique())partners=partners[(partners!=taxon)&(partners!="medium")]ints=[]forpinpartners:fluxes=ex[ex.taxon.isin((taxon,p))]ints.append(fluxes.groupby("metabolite").apply(lambdadf:_metabolite_interaction(df,taxon,p)).reset_index())ints=pd.concat([iforiinintsifiisnotNone])ints["sample_id"]=sample_idreturnints
[docs]def_interact(args:List)->pd.DataFrame:"""Quantify interactions of a focal taxon with other taxa."""results,taxon=argsex=results.exchanges[results.exchanges.taxon!="medium"]ints=(ex.groupby("sample_id").apply(lambdadf:sample_interactions(df,df.name,taxon)).reset_index(drop=True).drop(["level_1","index"],axis=1,errors="ignore").merge(results.annotations,on="metabolite"))returnints
[docs]definteractions(results:GrowthResults,taxa:Union[None,str,List[str]],threads:int=1,progress:bool=True,)->pd.DataFrame:"""Quantify interactions of a focal/reference taxon with other taxa. Arguments --------- results : GrowthResults The growth results to use. taxa : str, list of str, or None The focal taxa to use. Can be a single taxon, a list of taxa or None in which case all taxa are considered. Returns ------- pandas.DataFrame The mapped interactions between the focal taxon and all other taxa. """ifisinstance(taxa,str):return_interact([results,taxon_id(taxa,results.growth_rates)])eliftaxaisNone:taxa=results.growth_rates.taxon.unique()taxa=[taxon_id(t,results.growth_rates)fortintaxa]ints=pd.concat(workflow(_interact,[[results,t]fortintaxa],threads=threads,progress=progress))returnints