comparison COBRAxy/ras_to_bounds.py @ 216:b162b98f9de5 draft

Uploaded

215:5cc4a367ef70

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


def apply_ras_bounds(model, ras_row):
    """
    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.
    Returns:
        None
    """
    for 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).upper_bound=valMax
        if upper_bound==0 and lower_bound!=0:
            model.reactions.get_by_id(reaction).lower_bound=valMin
        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.

        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)
    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:
    """
    Generate reaction bounds for a metabolic model based on medium conditions and optional RAS adjustments.

        model.reactions.get_by_id(reaction).upper_bound = float(df_FVA.loc[reaction, "maximum"])

    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())
    else:
        model_new = model.copy()
        apply_ras_bounds(model_new, pd.Series([1]*len(rxns_ids), index=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



############################# main ###########################################

216:b162b98f9de5

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


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

    Args:
        bounds (pd.DataFrame): Model bounds.
        ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
    Returns:
        new_bounds (pd.DataFrame): integrated bounds.
    """
    new_bounds = bounds.copy()
    for reaction in ras_row.index:
        scaling_factor = ras_row[reaction]
        lower_bound=bounds.loc[reaction, "lower_bound"]
        upper_bound=bounds.loc[reaction, "upper_bound"]
        valMax=float((upper_bound)*scaling_factor)
        valMin=float((lower_bound)*scaling_factor)
        if upper_bound!=0 and lower_bound==0:
            new_bounds.loc[reaction, "upper_bound"] = valMax
        if upper_bound==0 and lower_bound!=0:
            new_bounds.loc[reaction, "lower_bound"] = valMin
        if upper_bound!=0 and lower_bound!=0:
            new_bounds.loc[reaction, "lower_bound"] = valMin
            new_bounds.loc[reaction, "upper_bound"] = valMax
    return new_bounds

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.

        output_folder (str): Folder path where the output CSV file will be saved.
    
    Returns:
        None
    """
    bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
    new_bounds = apply_ras_bounds(bounds, ras_row)
    new_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:
    """
    Generate reaction bounds for a metabolic model based on medium conditions and optional RAS adjustments.

        model.reactions.get_by_id(reaction).upper_bound = float(df_FVA.loc[reaction, "maximum"])

    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())
    else:
        bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
        newBounds = apply_ras_bounds(bounds, pd.Series([1]*len(rxns_ids), index=rxns_ids))
        newBounds.to_csv(output_folder + "bounds.csv", sep='\t', index=True)
    pass



############################# main ###########################################