comparison COBRAxy/ras_to_bounds_beta.py @ 456:a6e45049c1b9 draft

Uploaded

455:4e2bc80764b6

import argparse
import utils.general_utils as utils
from typing import Optional, Dict, Set, List, Tuple, Union
import os
import numpy as np
import pandas as pd
import cobra
from cobra import Model, Reaction, Metabolite
import re
import sys
import csv
from joblib import Parallel, delayed, cpu_count
import utils.rule_parsing  as rulesUtils
import utils.reaction_parsing as reactionUtils
import utils.model_utils as modelUtils

################################# process args ###############################
def process_args(args :List[str] = None) -> argparse.Namespace:
    """

            df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "Rule"])
            df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Reaction"])
            df_bounds = bounds.reset_index().rename(columns = {"index": "ReactionID"})
            df_medium = medium.rename(columns = {"reaction": "ReactionID"})
            df_medium["InMedium"] = True # flag per indicare la presenza nel medium

            merged = df_reactions.merge(df_rules, on = "ReactionID", how = "outer")
            merged = merged.merge(df_bounds, on = "ReactionID", how = "outer")
            
            # Add compartments only if they exist and model name is ENGRO2

    # Save model if requested
    if save_models:
        modified_model = apply_bounds_to_model(model, new_bounds)
        save_model(modified_model, cellName, save_models_path, save_models_format)
    
    pass

def generate_bounds_model(model: cobra.Model, ras=None, output_folder='output/', save_models=False, save_models_path='saved_models/', save_models_format='csv') -> pd.DataFrame:
    """
    Generate reaction bounds for a metabolic model based on medium conditions and optional RAS adjustments.
    

            cellName, ras_row, model, rxns_ids, output_folder, 
            save_models, save_models_path, save_models_format
        ) for cellName, ras_row in ras.iterrows())
    else:
        raise ValueError("RAS DataFrame is None. Cannot generate bounds without RAS data.")
    pass

############################# main ###########################################
def main(args:List[str] = None) -> None:
    """
    Initializes everything and sets the program in motion based on the fronted input arguments.

    Returns:
        None
    """
    if not os.path.exists('ras_to_bounds'):

    ras_file_names = ARGS.name.split(",")
    if len(ras_file_names) != len(set(ras_file_names)):
        error_message = "Duplicated file names in the uploaded RAS matrices."
        warning(error_message)
        raise ValueError(error_message)
        pass
    ras_class_names = []
    for file in ras_file_names:
        ras_class_names.append(file.rsplit(".", 1)[0])
    ras_list = []
    class_assignments = pd.DataFrame(columns=["Patient_ID", "Class"])

        ras.replace("None", None, inplace=True)
        ras.set_index("Reactions", drop=True, inplace=True)
        ras = ras.T
        ras = ras.astype(float)
        if(len(ras_file_list)>1):
            #append class name to patient id (dataframe index)
            ras.index = [f"{idx}_{ras_class_name}" for idx in ras.index]
        else:
            ras.index = [f"{idx}" for idx in ras.index]
        ras_list.append(ras)
        for patient_id in ras.index:
            class_assignments.loc[class_assignments.shape[0]] = [patient_id, ras_class_name]
    
        
        # Concatenate all ras DataFrames into a single DataFrame
        ras_combined = pd.concat(ras_list, axis=0)
        # Normalize the RAS values by max RAS
        ras_combined = ras_combined.div(ras_combined.max(axis=0))
        ras_combined.dropna(axis=1, how='all', inplace=True)

    model = modelUtils.build_cobra_model_from_csv(ARGS.model_upload)

    validation = modelUtils.validate_model(model)

    print("\n=== VALIDAZIONE MODELLO ===")
    for key, value in validation.items():
        print(f"{key}: {value}")


    generate_bounds_model(model, ras=ras_combined, output_folder=ARGS.output_path,
                    save_models=ARGS.save_models, save_models_path=ARGS.save_models_path,
                    save_models_format=ARGS.save_models_format)
    class_assignments.to_csv(ARGS.cell_class, sep='\t', index=False)


    pass
        
##############################################################################
if __name__ == "__main__":
    main()

456:a6e45049c1b9

"""
Apply RAS-based scaling to reaction bounds and optionally save updated models.

Workflow:
- Read one or more RAS matrices (patients/samples x reactions)
- Normalize and merge them, optionally adding class suffixes to sample IDs
- Build a COBRA model from a tabular CSV
- Run FVA to initialize bounds, then scale per-sample based on RAS values
- Save bounds per sample and optionally export updated models in chosen formats
"""
import argparse
import utils.general_utils as utils
from typing import Optional, Dict, Set, List, Tuple, Union
import os
import numpy as np
import pandas as pd
import cobra
from cobra import Model
import sys
from joblib import Parallel, delayed, cpu_count
import utils.model_utils as modelUtils

################################# process args ###############################
def process_args(args :List[str] = None) -> argparse.Namespace:
    """

            df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "Rule"])
            df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Reaction"])
            df_bounds = bounds.reset_index().rename(columns = {"index": "ReactionID"})
            df_medium = medium.rename(columns = {"reaction": "ReactionID"})
            df_medium["InMedium"] = True

            merged = df_reactions.merge(df_rules, on = "ReactionID", how = "outer")
            merged = merged.merge(df_bounds, on = "ReactionID", how = "outer")
            
            # Add compartments only if they exist and model name is ENGRO2

    # Save model if requested
    if save_models:
        modified_model = apply_bounds_to_model(model, new_bounds)
        save_model(modified_model, cellName, save_models_path, save_models_format)
    
    return

def generate_bounds_model(model: cobra.Model, ras=None, output_folder='output/', save_models=False, save_models_path='saved_models/', save_models_format='csv') -> pd.DataFrame:
    """
    Generate reaction bounds for a metabolic model based on medium conditions and optional RAS adjustments.
    

            cellName, ras_row, model, rxns_ids, output_folder, 
            save_models, save_models_path, save_models_format
        ) for cellName, ras_row in ras.iterrows())
    else:
        raise ValueError("RAS DataFrame is None. Cannot generate bounds without RAS data.")
    return

############################# main ###########################################
def main(args:List[str] = None) -> None:
    """
    Initialize and execute RAS-to-bounds pipeline based on the frontend input arguments.

    Returns:
        None
    """
    if not os.path.exists('ras_to_bounds'):

    ras_file_names = ARGS.name.split(",")
    if len(ras_file_names) != len(set(ras_file_names)):
        error_message = "Duplicated file names in the uploaded RAS matrices."
        warning(error_message)
        raise ValueError(error_message)
        
    ras_class_names = []
    for file in ras_file_names:
        ras_class_names.append(file.rsplit(".", 1)[0])
    ras_list = []
    class_assignments = pd.DataFrame(columns=["Patient_ID", "Class"])

        ras.replace("None", None, inplace=True)
        ras.set_index("Reactions", drop=True, inplace=True)
        ras = ras.T
        ras = ras.astype(float)
        if(len(ras_file_list)>1):
            # Append class name to patient id (DataFrame index)
            ras.index = [f"{idx}_{ras_class_name}" for idx in ras.index]
        else:
            ras.index = [f"{idx}" for idx in ras.index]
        ras_list.append(ras)
        for patient_id in ras.index:
            class_assignments.loc[class_assignments.shape[0]] = [patient_id, ras_class_name]
    
        
    # Concatenate all RAS DataFrames into a single DataFrame
        ras_combined = pd.concat(ras_list, axis=0)
    # Normalize RAS values column-wise by max RAS
        ras_combined = ras_combined.div(ras_combined.max(axis=0))
        ras_combined.dropna(axis=1, how='all', inplace=True)

    model = modelUtils.build_cobra_model_from_csv(ARGS.model_upload)

    validation = modelUtils.validate_model(model)

    print("\n=== MODEL VALIDATION ===")
    for key, value in validation.items():
        print(f"{key}: {value}")


    generate_bounds_model(model, ras=ras_combined, output_folder=ARGS.output_path,
                    save_models=ARGS.save_models, save_models_path=ARGS.save_models_path,
                    save_models_format=ARGS.save_models_format)
    class_assignments.to_csv(ARGS.cell_class, sep='\t', index=False)


    return
        
##############################################################################
if __name__ == "__main__":
    main()