|
| 1 | +#!/usr/bin/env python |
| 2 | +""" Provides functions / connections to after tools for easier access and usage. |
| 3 | +""" |
| 4 | + |
| 5 | +__author__ = "Famke Baeuerle und Carolin Brune" |
| 6 | + |
| 7 | +################################################################################ |
| 8 | +# requirements |
| 9 | +################################################################################ |
| 10 | + |
| 11 | +import cobra |
| 12 | +import json |
| 13 | +import memote |
| 14 | +import tempfile |
| 15 | +import time |
| 16 | +import warnings |
| 17 | + |
| 18 | +from BOFdat import step1 |
| 19 | +from BOFdat import step2 |
| 20 | +from BOFdat.util import update |
| 21 | +from BOFdat.util.update import determine_coefficients |
| 22 | + |
| 23 | +from MCC import MassChargeCuration |
| 24 | +from pathlib import Path |
| 25 | +from typing import Literal |
| 26 | + |
| 27 | +from memote.support import consistency |
| 28 | +# needed by memote.support.consistency |
| 29 | +from memote.support import consistency_helpers as con_helpers |
| 30 | + |
| 31 | +from ..curation.biomass import test_biomass_presence |
| 32 | + |
| 33 | +# note: |
| 34 | +# for BOFdat to run correctly, you need to change 'solution.f' to 'solution.objective_value' |
| 35 | +# in the coenzymes_and_ions.py file of BOFdat |
| 36 | + |
| 37 | +################################################################################ |
| 38 | +# variables |
| 39 | +################################################################################ |
| 40 | + |
| 41 | +################################################################################ |
| 42 | +# functions |
| 43 | +################################################################################ |
| 44 | + |
| 45 | +# BOFdat |
| 46 | +# ------ |
| 47 | + |
| 48 | +def adjust_BOF(genome:str, model_file:str, model:cobra.Model, dna_weight_fraction:float, weight_frac:float) -> str: |
| 49 | + """Adjust the model's BOF using BOFdat. Currently implemented are step 1 |
| 50 | + DNA coefficients and step 2. |
| 51 | +
|
| 52 | + Args: |
| 53 | + - genome (str): |
| 54 | + Path to the genome (e.g. .fna) FASTA file. |
| 55 | + - model_file (str): |
| 56 | + Path to the sbml (.xml) file of the model. |
| 57 | + - model (cobra.Model): |
| 58 | + The genome-scale metabolic model (from the string above), loaded with COBRApy. |
| 59 | + - dna_weight_fraction (float): |
| 60 | + DNA weight fraction for BOF step 1. |
| 61 | + - weight_frac (float): |
| 62 | + Weight fraction for the second step of BOFdat (enzymes and ions) |
| 63 | +
|
| 64 | + Returns: |
| 65 | + str: |
| 66 | + The updated BOF reaction as a reaction string. |
| 67 | + """ |
| 68 | + |
| 69 | + |
| 70 | + # BOFdat step 1: |
| 71 | + # -------------- |
| 72 | + # dna coefficients |
| 73 | + dna_coefficients = step1.generate_dna_coefficients(genome,model_file,DNA_WEIGHT_FRACTION=dna_weight_fraction) |
| 74 | + bd_step1 = {} |
| 75 | + for m in dna_coefficients: |
| 76 | + bd_step1[m.id] = dna_coefficients[m] |
| 77 | + |
| 78 | + # ........................... |
| 79 | + # @TODO |
| 80 | + # if time permits or needed, options for more coefficients can be added |
| 81 | + # ........................... |
| 82 | + |
| 83 | + # BOFdat step 2: |
| 84 | + # -------------- |
| 85 | + # find inorganic ions |
| 86 | + selected_metabolites = step2.find_coenzymes_and_ions(model_file) |
| 87 | + # determine coefficients |
| 88 | + bd_step2 = determine_coefficients(selected_metabolites,model,weight_frac) |
| 89 | + bd_step2.update(bd_step1) |
| 90 | + |
| 91 | + # update BOF |
| 92 | + # ---------- |
| 93 | + # retrieve previously used BOF |
| 94 | + growth_func_list = test_biomass_presence(model) |
| 95 | + if len(growth_func_list) == 1: |
| 96 | + objective_list = model.reactions.get_by_id(growth_func_list[0]).reaction.split(' ') |
| 97 | + elif len(growth_func_list) > 1: |
| 98 | + mes = f'Multiple BOFs found. Using {growth_func_list[0]} for BOF adjustment.' |
| 99 | + warnings.warn(mes,category=UserWarning) |
| 100 | + objective_list = model.reactions.get_by_id(growth_func_list[0]).reaction.split(' ') |
| 101 | + # else not needed, as if there is no BOF, a new one will be created |
| 102 | + |
| 103 | + # ............................................................... |
| 104 | + objective_reactant = {} |
| 105 | + objective_product = {} |
| 106 | + product = False |
| 107 | + # get reactants, product and factors from equation |
| 108 | + for s in objective_list: |
| 109 | + if s == '+': |
| 110 | + continue |
| 111 | + elif '>' in s: |
| 112 | + product = True |
| 113 | + elif '.' in s: |
| 114 | + factor = s |
| 115 | + else: |
| 116 | + if product: |
| 117 | + objective_product[s] = factor |
| 118 | + else: |
| 119 | + objective_reactant[s] = factor |
| 120 | + |
| 121 | + # update BOF information with data from BOFdat |
| 122 | + for m in bd_step2: |
| 123 | + if bd_step2[m] < 0: |
| 124 | + objective_reactant[m] = bd_step2[m]*(-1) |
| 125 | + else: |
| 126 | + objective_product[m] = bd_step2[m] |
| 127 | + |
| 128 | + # create new objective function |
| 129 | + new_objective = ' + '.join('{} {}'.format(value, key) for key, value in objective_reactant.items()) |
| 130 | + new_objective += ' --> ' |
| 131 | + new_objective += ' + '.join('{} {}'.format(value, key) for key, value in objective_product.items()) |
| 132 | + |
| 133 | + return new_objective |
| 134 | + |
| 135 | + |
| 136 | + |
| 137 | +# MCC - MassChargeCuration |
| 138 | +# ------------------------ |
| 139 | + |
| 140 | +def perform_mcc(model: cobra.Model, dir: str, apply:bool = True) -> cobra.Model: |
| 141 | + """Run the MassChargeCuration toll on the model and optionally directly apply |
| 142 | + the solution. |
| 143 | +
|
| 144 | + Args: |
| 145 | + - model (cobra.Model): |
| 146 | + The model to use the tool on. |
| 147 | + - dir (str): |
| 148 | + Path of the directory to save MCC output in. |
| 149 | + - apply (bool, optional): |
| 150 | + If True, model is directly updated with the results. |
| 151 | + Defaults to True. |
| 152 | +
|
| 153 | + Returns: |
| 154 | + cobra.Model: |
| 155 | + The model (updated or not) |
| 156 | + """ |
| 157 | + |
| 158 | + # make temporary directory to save files for MCC in |
| 159 | + with tempfile.TemporaryDirectory() as temp: |
| 160 | + |
| 161 | + # use MCC |
| 162 | + if apply: |
| 163 | + # update model |
| 164 | + balancer = MassChargeCuration(model, update_ids = False, data_path=temp) |
| 165 | + else: |
| 166 | + # do not change original model |
| 167 | + model_copy = model.copy() |
| 168 | + balancer = MassChargeCuration(model_copy, update_ids = False, data_path=temp) |
| 169 | + |
| 170 | + # save reports |
| 171 | + balancer.generate_reaction_report(Path(dir,model.id+'_mcc_reactions')) |
| 172 | + balancer.generate_metabolite_report(Path(dir,model.id+'_mcc_metabolites')) |
| 173 | + balancer.generate_visual_report(Path(dir,model.id+'_mcc_visual')) |
| 174 | + |
| 175 | + return model |
| 176 | + |
| 177 | + |
| 178 | + |
| 179 | +# Memote |
| 180 | +# ------- |
| 181 | + |
| 182 | +def run_memote(model: cobra.Model, type:Literal['json','html']='html', |
| 183 | + return_res:bool=False, save_res:str|None=None, verbose:bool=False) -> dict|str|None: |
| 184 | + """Run the memote snapshot function on a given model loaded with COBRApy. |
| 185 | +
|
| 186 | + Args: |
| 187 | + - model (cobra.Model): |
| 188 | + The model loaded with COBRApy. |
| 189 | + - type (Literal['json','html'], optional): |
| 190 | + Type of report to produce. |
| 191 | + Can be 'html' or 'json'. |
| 192 | + Defaults to 'html'. |
| 193 | + - return_res (bool, optional): |
| 194 | + Option to return the result. |
| 195 | + Defaults to False. |
| 196 | + - save_res (str | None, optional): |
| 197 | + If given a path string, saves the report |
| 198 | + under the given path. Defaults to None. |
| 199 | + - verbose (bool, optional): |
| 200 | + Produce a more verbose ouput. |
| 201 | + Defaults to False. |
| 202 | +
|
| 203 | + Raises: |
| 204 | + ValueError: Unknown input for parameter type |
| 205 | +
|
| 206 | + Returns: |
| 207 | + (1) Case return_res = True and type = json |
| 208 | +
|
| 209 | + dict: The json dictionary. |
| 210 | +
|
| 211 | + (2) Case return_res = True and type = html |
| 212 | +
|
| 213 | + str: The html string. |
| 214 | +
|
| 215 | + (3) Case return_res = False |
| 216 | + |
| 217 | + None: no return |
| 218 | + """ |
| 219 | + |
| 220 | + # verbose output I |
| 221 | + if verbose: |
| 222 | + print('\n# -------------------\n# Analyse with MEMOTE\n# -------------------') |
| 223 | + start = time.time() |
| 224 | + |
| 225 | + # run memote |
| 226 | + ret, res = memote.suite.api.test_model(model, sbml_version=None, results=True, |
| 227 | + pytest_args=None, exclusive=None, skip=None, |
| 228 | + experimental=None, solver_timeout=10) |
| 229 | + |
| 230 | + # load depending on type |
| 231 | + match type: |
| 232 | + case 'html': |
| 233 | + snap = memote.suite.api.snapshot_report(res, html=True) |
| 234 | + result = snap |
| 235 | + case 'json': |
| 236 | + snap = memote.suite.api.snapshot_report(res, html=False) |
| 237 | + result = json.loads(snap) |
| 238 | + case _: |
| 239 | + message = f'Unknown input for parameter type: {type} ' |
| 240 | + raise ValueError(message) |
| 241 | + |
| 242 | + # option to save report |
| 243 | + if save_res: |
| 244 | + with open(save_res, 'w') as f: |
| 245 | + f.write(result) |
| 246 | + |
| 247 | + # verbose output II |
| 248 | + if verbose: |
| 249 | + end = time.time() |
| 250 | + print(F'\ttotal time: {end - start}s') |
| 251 | + |
| 252 | + # option to return report |
| 253 | + if return_res: |
| 254 | + return result |
| 255 | + |
| 256 | + |
| 257 | +def get_memote_score(memote_report: dict) -> float: |
| 258 | + """Extracts MEMOTE score from report |
| 259 | +
|
| 260 | + Args: |
| 261 | + - memote_report (dict): |
| 262 | + Output from run_memote. |
| 263 | +
|
| 264 | + Returns: |
| 265 | + float: |
| 266 | + MEMOTE score |
| 267 | + """ |
| 268 | + return memote_report['score']['total_score'] |
| 269 | + |
| 270 | + |
| 271 | + |
| 272 | + |
0 commit comments