comparison COBRAxy/custom_data_generator.py @ 403:05092b0cfca0 draft

Uploaded

402:ccccb731c953

    for reaction in model.reactions:
        bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound]
    return bounds


###############################- FILE SAVING -################################
def save_as_csv_filePath(data :dict, file_path :utils.FilePath, fieldNames :Tuple[str, str]) -> None:
    """
    Saves any dictionary-shaped data in a .csv file created at the given file_path as FilePath.

            raise utils.DataErr(ARGS.model, f"failed loading built-in model: {e}")

    # Determine final model name: explicit --name overrides, otherwise use the model id
    
    model_name = ARGS.name if ARGS.name else ARGS.model
    print(ARGS.name)
    print(model_name)
    print(ARGS.medium_selector)
    
    if ARGS.name == "ENGRO2" and ARGS.medium_selector != "Default":
        df_mediums = pd.read_csv(ARGS.tool_dir + "/local/medium/medium.csv", index_col = 0)
        ARGS.medium_selector = ARGS.medium_selector.replace("_", " ")
        medium = df_mediums[[ARGS.medium_selector]]

    # generate data
    rules = generate_rules(model, asParsed = False)
    reactions = generate_reactions(model, asParsed = False)
    bounds = generate_bounds(model)
    medium = get_medium(model)

    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")

    merged = merged.merge(df_medium, on = "ReactionID", how = "left")

    merged["InMedium"] = merged["InMedium"].fillna(False)

    merged = merged.sort_values(by = "InMedium", ascending = False)

403:05092b0cfca0

    for reaction in model.reactions:
        bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound]
    return bounds



def generate_compartments(model: cobra.Model) -> pd.DataFrame:
    """
    Generates a DataFrame containing compartment information for each reaction.
    Creates columns for each compartment position (Compartment_1, Compartment_2, etc.)
    
    Args:
        model: the COBRA model to extract compartment data from.
        
    Returns:
        pd.DataFrame: DataFrame with ReactionID and compartment columns
    """
    compartment_data = []
    
    # First pass: determine the maximum number of compartments any reaction has
    max_compartments = 0
    reaction_compartments = {}
    
    for reaction in model.reactions:
        # Get unique compartments from all metabolites in the reaction
        if type(reaction.annotation['pathways']) == list:
            reaction_compartments[reaction.id] = reaction.annotation['pathways']
            max_compartments = max(max_compartments, len(reaction.annotation['pathways']))
        else:
            reaction_compartments[reaction.id] = [reaction.annotation['pathways']]
    
    # Create column names for compartments
    compartment_columns = [f"Compartment_{i+1}" for i in range(max_compartments)]

    # Second pass: create the data
    for reaction_id, compartments in reaction_compartments.items():
        row = {"ReactionID": reaction_id}
        
        # Fill compartment columns
        for i in range(max_compartments):
            col_name = compartment_columns[i]
            if i < len(compartments):
                row[col_name] = compartments[i]
                
            else:
                row[col_name] = None  # or "" if you prefer empty strings
                
        compartment_data.append(row)
    
    return pd.DataFrame(compartment_data)


###############################- FILE SAVING -################################
def save_as_csv_filePath(data :dict, file_path :utils.FilePath, fieldNames :Tuple[str, str]) -> None:
    """
    Saves any dictionary-shaped data in a .csv file created at the given file_path as FilePath.

            raise utils.DataErr(ARGS.model, f"failed loading built-in model: {e}")

    # Determine final model name: explicit --name overrides, otherwise use the model id
    
    model_name = ARGS.name if ARGS.name else ARGS.model
    
    if ARGS.name == "ENGRO2" and ARGS.medium_selector != "Default":
        df_mediums = pd.read_csv(ARGS.tool_dir + "/local/medium/medium.csv", index_col = 0)
        ARGS.medium_selector = ARGS.medium_selector.replace("_", " ")
        medium = df_mediums[[ARGS.medium_selector]]

    # generate data
    rules = generate_rules(model, asParsed = False)
    reactions = generate_reactions(model, asParsed = False)
    bounds = generate_bounds(model)
    medium = get_medium(model)
    compartments = generate_compartments(model)

    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")
    merged = merged.merge(compartments, on = "ReactionID", how = "outer")
    merged = merged.merge(df_medium, on = "ReactionID", how = "left")

    merged["InMedium"] = merged["InMedium"].fillna(False)

    merged = merged.sort_values(by = "InMedium", ascending = False)