comparison COBRAxy/ras_to_bounds.py @ 113:1961a091484c draft

Uploaded

112:4ea14da7043b

    if len(dataset.columns) < 2:
        sys.exit('Execution aborted: wrong format of ' + name + '\n')
    return dataset


def apply_ras_bounds(model, ras_row, rxns_ids):
    """
    Adjust the bounds of reactions in the model based on RAS values.

    Args:
        model (cobra.Model): The metabolic model to be modified.
        ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
        rxns_ids (list of str): List of reaction IDs to which the scaling factors will be applied.
    
    Returns:
        None
    """
    for reaction in rxns_ids:
        if reaction in ras_row.index:
            scaling_factor = ras_row[reaction]
            lower_bound=model.reactions.get_by_id(reaction).lower_bound
            upper_bound=model.reactions.get_by_id(reaction).upper_bound
            valMax=float((upper_bound)*scaling_factor)
            valMin=float((lower_bound)*scaling_factor)

            if upper_bound!=0 and lower_bound!=0:
                model.reactions.get_by_id(reaction).lower_bound=valMin
                model.reactions.get_by_id(reaction).upper_bound=valMax
    pass

def process_ras_cell(cellName, ras_row, model, rxns_ids, output_folder):
    """
    Process a single RAS cell, apply bounds, and save the bounds to a CSV file.

    Args:
        cellName (str): The name of the RAS cell (used for naming the output file).
        ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
        model (cobra.Model): The metabolic model to be modified.
        rxns_ids (list of str): List of reaction IDs to which the scaling factors will be applied.
        output_folder (str): Folder path where the output CSV file will be saved.
    
    Returns:
        None
    """
    model_new = model.copy()
    apply_ras_bounds(model_new, ras_row, rxns_ids)
    bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model_new.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
    bounds.to_csv(output_folder + cellName + ".csv", sep='\t', index=True)
    pass

def generate_bounds(model: cobra.Model, medium: dict, ras=None, output_folder='output/') -> pd.DataFrame:

        if value is not None:
            if(reaction != "EX_thbpt_e" and reaction != "EX_lac__L_e"):
                model.reactions.get_by_id(reaction).lower_bound = -float(value)
            if(reaction == "EX_lac__L_e"):
                model.reactions.get_by_id(reaction).lower_bound = float(0.0)
            
    # Perform Flux Variability Analysis (FVA)
    df_FVA = cobra.flux_analysis.flux_variability_analysis(model, fraction_of_optimum=0, processes=1).round(8)
    
    # Set FVA bounds
    for reaction in rxns_ids:
        model.reactions.get_by_id(reaction).lower_bound = float(df_FVA.loc[reaction, "minimum"])
        model.reactions.get_by_id(reaction).upper_bound = float(df_FVA.loc[reaction, "maximum"])

    warning(str(model.reactions.get_by_id("EX_Lcystin_e").lower_bound))

    if ras is not None:
        #Parallel(n_jobs=cpu_count())(delayed(process_ras_cell)(cellName, ras_row, model, rxns_ids, output_folder) for cellName, ras_row in ras.iterrows())
        for cellName, ras_row in ras.iterrows():
            process_ras_cell(cellName, ras_row, model, rxns_ids, output_folder)
            break #just one cell for testing
    else:
        model_new = model.copy()
        apply_ras_bounds(model_new, pd.Series([1]*len(rxns_ids), index=rxns_ids), rxns_ids)
        bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model_new.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
        bounds.to_csv(output_folder + "bounds.csv", sep='\t', index=True)
    pass

113:1961a091484c

    if len(dataset.columns) < 2:
        sys.exit('Execution aborted: wrong format of ' + name + '\n')
    return dataset


def apply_ras_bounds(model, ras_row, rxns_ids, mediumRxns_ids):
    """
    Adjust the bounds of reactions in the model based on RAS values.

    Args:
        model (cobra.Model): The metabolic model to be modified.
        ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
        rxns_ids (list of str): List of reaction IDs to which the scaling factors will be applied.
        mediumRxns_ids (list of str): List of reaction IDs in the medium. Their RAS is set to zero, but they are already set in the model.
    Returns:
        None
    """
    for reaction in rxns_ids:
        if reaction in ras_row.index and ras_row[reaction] not in mediumRxns_ids:
            scaling_factor = ras_row[reaction]
            lower_bound=model.reactions.get_by_id(reaction).lower_bound
            upper_bound=model.reactions.get_by_id(reaction).upper_bound
            valMax=float((upper_bound)*scaling_factor)
            valMin=float((lower_bound)*scaling_factor)

            if upper_bound!=0 and lower_bound!=0:
                model.reactions.get_by_id(reaction).lower_bound=valMin
                model.reactions.get_by_id(reaction).upper_bound=valMax
    pass

def process_ras_cell(cellName, ras_row, model, rxns_ids, mediumRxns_ids, output_folder):
    """
    Process a single RAS cell, apply bounds, and save the bounds to a CSV file.

    Args:
        cellName (str): The name of the RAS cell (used for naming the output file).
        ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
        model (cobra.Model): The metabolic model to be modified.
        rxns_ids (list of str): List of reaction IDs to which the scaling factors will be applied.
        mediumRxns_ids (list of str): List of reaction IDs in the medium. Their RAS is set to zero, but they are already set in the model.
        output_folder (str): Folder path where the output CSV file will be saved.
    
    Returns:
        None
    """
    model_new = model.copy()
    apply_ras_bounds(model_new, ras_row, rxns_ids, mediumRxns_ids)
    bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model_new.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
    bounds.to_csv(output_folder + cellName + ".csv", sep='\t', index=True)
    pass

def generate_bounds(model: cobra.Model, medium: dict, ras=None, output_folder='output/') -> pd.DataFrame:

        if value is not None:
            if(reaction != "EX_thbpt_e" and reaction != "EX_lac__L_e"):
                model.reactions.get_by_id(reaction).lower_bound = -float(value)
            if(reaction == "EX_lac__L_e"):
                model.reactions.get_by_id(reaction).lower_bound = float(0.0)

    mediumRxns_ids = medium.keys()
            
    # Perform Flux Variability Analysis (FVA)
    df_FVA = cobra.flux_analysis.flux_variability_analysis(model, fraction_of_optimum=0, processes=1).round(8)
    
    # Set FVA bounds
    for reaction in rxns_ids:
        model.reactions.get_by_id(reaction).lower_bound = float(df_FVA.loc[reaction, "minimum"])
        model.reactions.get_by_id(reaction).upper_bound = float(df_FVA.loc[reaction, "maximum"])

    #warning(str(model.reactions.get_by_id("EX_Lcystin_e").lower_bound))

    if ras is not None:
        #Parallel(n_jobs=cpu_count())(delayed(process_ras_cell)(cellName, ras_row, model, rxns_ids, output_folder) for cellName, ras_row in ras.iterrows())
        for cellName, ras_row in ras.iterrows():
            process_ras_cell(cellName, ras_row, model, rxns_ids, mediumRxns_ids, output_folder)
            break #just one cell for testing
    else:
        model_new = model.copy()
        apply_ras_bounds(model_new, pd.Series([1]*len(rxns_ids), index=rxns_ids), rxns_ids, mediumRxns_ids)
        bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model_new.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
        bounds.to_csv(output_folder + "bounds.csv", sep='\t', index=True)
    pass