Source code for cameo.core.solution

# -*- coding: utf-8 -*-
# Copyright 2013 Novo Nordisk Foundation Center for Biosustainability, DTU.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import absolute_import, print_function

import datetime
import logging
import time
from collections import OrderedDict

import cobra
from pandas import DataFrame, Series

import cameo
from cameo.exceptions import UndefinedSolution

logger = logging.getLogger(__name__)


[docs]class SolutionBase(object): def __new__(cls, *args, **kwargs): # this is a cobrapy compatibility hack if len(args) == 1 and not isinstance(args[0], cameo.core.solver_based_model.SolverBasedModel): cobrapy_solution = super(SolutionBase, cls).__new__(cobra.core.Solution) cobrapy_solution.__init__(*args, **kwargs) return cobrapy_solution else: return super(SolutionBase, cls).__new__(cls) def __init__(self, model): self.model = model self._x = None self._y = None self._x_dict = None self._y_dict = None @property def data_frame(self): return DataFrame({'fluxes': Series(self.fluxes), 'reduced_costs': Series(self.reduced_costs)}) def __str__(self): """A pandas DataFrame representation of the solution. Returns ------- pandas.DataFrame """ return str(self.data_frame) def _repr_html_(self): return self.data_frame._repr_html_()
[docs] def as_cobrapy_solution(self): """Convert into a cobrapy Solution. Returns ------- cobra.core.Solution.Solution """ return Solution(self.f, x=self.x, x_dict=self.x_dict, y=self.y, y_dict=self.y_dict, the_solver=None, the_time=0, status=self.status)
[docs] def get_primal_by_id(self, reaction_id): """Return a flux/primal value for a reaction. Parameters ---------- reaction_id : str A reaction ID. """ return self.x_dict[reaction_id]
@property def x_dict(self): if self._x_dict is None: return self.fluxes else: return self._x_dict @x_dict.setter def x_dict(self, value): self._x_dict = value @property def x(self): if self._x is None: return self.fluxes.values() else: return self._x @x.setter def x(self, value): self._x = value @property def y_dict(self): if self._y_dict is None: return self.reduced_costs else: return self._y_dict @y_dict.setter def y_dict(self, value): self._y_dict = value @property def y(self): if self._y is None: return self.reduced_costs.values() else: return self._y @y.setter def y(self, value): self._y = value @property def objective_value(self): return self.f
[docs]class Solution(SolutionBase): """This class mimicks the cobrapy Solution class. Attributes ---------- fluxes : OrderedDict A dictionary of flux values. reduced_costs : OrderedDict A dictionary of reduced costs. Notes ----- See also documentation for cobra.core.Solution.Solution for an extensive list of inherited attributes. """ def __init__(self, model, *args, **kwargs): """ Parameters ---------- model : SolverBasedModel """ super(Solution, self).__init__(model, *args, **kwargs) self.f = model.solver.objective.value self.fluxes = OrderedDict() self.shadow_prices = model.solver.shadow_prices self.reduced_costs = OrderedDict() self._primal_values = model.solver.primal_values self._reduced_values = model.solver.reduced_costs for reaction in model.reactions: self.fluxes[reaction.id] = self._primal_values[reaction._get_forward_id()] - self._primal_values[ reaction._get_reverse_id()] self.reduced_costs[reaction.id] = self._reduced_values[reaction._get_forward_id()] - self._reduced_values[ reaction._get_reverse_id()] self.status = model.solver.status self._reaction_ids = [r.id for r in self.model.reactions] self._metabolite_ids = [m.id for m in self.model.metabolites] def __dir__(self): # Hide 'cobrapy' attributes and methods from user. fields = sorted(dir(type(self)) + list(self.__dict__.keys())) fields.remove('x') fields.remove('y') fields.remove('x_dict') fields.remove('y_dict') return fields def _repr_html_(self): return "%s: %f" % (self.model.objective.expression, self.f)
[docs]class LazySolution(SolutionBase): """This class implements a lazy evaluating version of the cobrapy Solution class. Attributes ---------- model : SolverBasedModel fluxes : OrderedDict A dictionary of flux values. reduced_costs : OrderedDict A dictionary of reduced costs. Notes ----- See also documentation for cobra.core.Solution.Solution for an extensive list of inherited attributes. """ def __init__(self, model, *args, **kwargs): """ Parameters ---------- model : SolverBasedModel """ super(LazySolution, self).__init__(model, *args, **kwargs) if self.model._timestamp_last_optimization is not None: self._time_stamp = self.model._timestamp_last_optimization else: self._time_stamp = time.time() self._f = None self._primal_values = None self._reduced_values = None def _check_freshness(self): """Raises an exceptions if the solution might have become invalid due to re-optimization of the attached model. Raises ------ UndefinedSolution If solution has become invalid. """ if self._time_stamp != self.model._timestamp_last_optimization: def timestamp_formatter(timestamp): datetime.datetime.fromtimestamp(timestamp).strftime( "%Y-%m-%d %H:%M:%S:%f") raise UndefinedSolution( 'The solution (captured around %s) has become invalid as the model has been ' 're-optimized recently (%s).' % ( timestamp_formatter(self._time_stamp), timestamp_formatter(self.model._timestamp_last_optimization)) ) @property def status(self): self._check_freshness() return self.model.solver.status @property def f(self): self._check_freshness() if self._f is None: return self.model.solver.objective.value else: return self._f @f.setter def f(self, value): self._f = value @property def fluxes(self): self._check_freshness() primal_values = self.model.solver.primal_values fluxes = OrderedDict() for reaction in self.model.reactions: fluxes[reaction.id] = primal_values[reaction._get_forward_id()] - primal_values[reaction._get_reverse_id()] return fluxes @property def reduced_costs(self): self._check_freshness() reduced_values = self.model.solver.reduced_costs reduced_costs = OrderedDict() for reaction in self.model.reactions: reduced_costs[reaction.id] = reduced_values[reaction._get_forward_id()] - reduced_values[ reaction._get_reverse_id()] return reduced_costs @property def shadow_prices(self): self._check_freshness() return self.model.solver.shadow_prices
[docs] def get_primal_by_id(self, reaction_id): """Return a flux/primal value for a reaction. Parameters ---------- reaction_id : str A reaction ID. """ self._check_freshness() return self.model.reactions.get_by_id(reaction_id).flux
def __dir__(self): # Hide 'cobrapy' attributes and methods from user. fields = sorted(dir(type(self)) + list(self.__dict__.keys())) fields.remove('x') fields.remove('y') fields.remove('x_dict') fields.remove('y_dict') return fields