Commit d6f252fe authored by Moritz E. Beber's avatar Moritz E. Beber

style: apply QA

parent ea7a739e
#!/usr/bin/env python
import versioneer
from setuptools import setup
import versioneer
# Most arguments are set in the `setup.cfg`.
setup(
version=versioneer.get_version(),
cmdclass=versioneer.get_cmdclass(),
)
setup(version=versioneer.get_version(), cmdclass=versioneer.get_cmdclass())
......@@ -33,8 +33,14 @@ import pkg_resources
from numpy import log
from . import (
Q_, Compound, ccache, default_conc_lb, default_conc_ub,
standard_concentration, ureg)
Q_,
Compound,
ccache,
default_conc_lb,
default_conc_ub,
standard_concentration,
ureg,
)
class BaseBounds(object):
......@@ -60,10 +66,7 @@ class BaseBounds(object):
"""
raise NotImplementedError
def GetLowerBounds(
self,
compounds: Iterable[Compound]
) -> Iterable[float]:
def GetLowerBounds(self, compounds: Iterable[Compound]) -> Iterable[float]:
"""Get the bounds for a set of keys in order.
:param compounds: an iterable of Compounds
......@@ -71,10 +74,7 @@ class BaseBounds(object):
"""
return map(self.GetLowerBound, compounds)
def GetUpperBounds(
self,
compounds: Iterable[Compound]
) -> Iterable[float]:
def GetUpperBounds(self, compounds: Iterable[Compound]) -> Iterable[float]:
"""Get the bounds for a set of keys in order.
:param compounds: an iterable of Compounds
......@@ -91,8 +91,7 @@ class BaseBounds(object):
return self.GetLowerBound(compound), self.GetUpperBound(compound)
def GetBounds(
self,
compounds: Iterable[Compound]
self, compounds: Iterable[Compound]
) -> Tuple[Iterable[float], Iterable[float]]:
"""Get the bounds for a set of compounds.
......@@ -114,8 +113,7 @@ class BaseBounds(object):
return log(b / standard_concentration)
def GetLnBounds(
self,
compounds: Iterable[Compound]
self, compounds: Iterable[Compound]
) -> Tuple[Iterable[float], Iterable[float]]:
"""Get the log-bounds for a set of compounds.
......@@ -127,8 +125,7 @@ class BaseBounds(object):
return map(self.conc2ln_conc, lbs), map(self.conc2ln_conc, ubs)
def GetLnLowerBounds(
self,
compounds: Iterable[Compound]
self, compounds: Iterable[Compound]
) -> Iterable[float]:
"""Get the log lower bounds for a set of compounds.
......@@ -139,8 +136,7 @@ class BaseBounds(object):
return map(self.conc2ln_conc, lbs)
def GetLnUpperBounds(
self,
compounds: Iterable[Compound]
self, compounds: Iterable[Compound]
) -> Iterable[float]:
"""Get the log upper bounds for a set of compounds.
......@@ -150,7 +146,7 @@ class BaseBounds(object):
ubs = self.GetUpperBounds(compounds)
return map(self.conc2ln_conc, ubs)
@ureg.check(None, None, '[concentration]', '[concentration]')
@ureg.check(None, None, "[concentration]", "[concentration]")
def SetBounds(self, compound: Compound, lb: float, ub: float):
"""Set bounds for a specific key.
......@@ -171,12 +167,14 @@ class Bounds(BaseBounds):
Allows for defaults
"""
@ureg.check(None, None, None, '[concentration]', '[concentration]')
def __init__(self,
lower_bounds: Dict[Compound, float] = None,
upper_bounds: Dict[Compound, float] = None,
default_lb: float = default_conc_lb,
default_ub: float = default_conc_ub) -> object:
@ureg.check(None, None, None, "[concentration]", "[concentration]")
def __init__(
self,
lower_bounds: Dict[Compound, float] = None,
upper_bounds: Dict[Compound, float] = None,
default_lb: float = default_conc_lb,
default_ub: float = default_conc_ub,
) -> object:
"""Initialize the bounds object.
Args:
......@@ -188,19 +186,21 @@ class Bounds(BaseBounds):
self.lower_bounds = lower_bounds or dict()
self.upper_bounds = upper_bounds or dict()
for b in self.lower_bounds.values():
assert b.check('[concentration]')
assert b.check("[concentration]")
for b in self.upper_bounds.values():
assert b.check('[concentration]')
assert b.check("[concentration]")
self.default_lb = default_lb
self.default_ub = default_ub
@classmethod
@ureg.check(None, None, '[concentration]', '[concentration]')
def from_csv(cls,
f: TextIO,
default_lb: float = default_conc_lb,
default_ub: float = default_conc_ub) -> object:
@ureg.check(None, None, "[concentration]", "[concentration]")
def from_csv(
cls,
f: TextIO,
default_lb: float = default_conc_lb,
default_ub: float = default_conc_ub,
) -> object:
"""Read bounds from a .csv file.
Assume that all the concentrations are given in units of molar.
......@@ -233,17 +233,17 @@ class Bounds(BaseBounds):
data = []
for c in self.lower_bounds.keys():
data.append((str(c), self.GetLowerBound(c), self.GetUpperBound(c)))
compound_df = pd.DataFrame(data=data, columns=['compound', 'lb', 'ub'])
compound_df = pd.DataFrame(data=data, columns=["compound", "lb", "ub"])
return compound_df
@classmethod
@ureg.check(None, None, None, '[concentration]', '[concentration]')
@ureg.check(None, None, None, "[concentration]", "[concentration]")
def from_dataframe(
cls,
df: pd.DataFrame,
bounds_unit: str = None,
default_lb: float = default_conc_lb,
default_ub: float = default_conc_ub
default_ub: float = default_conc_ub,
) -> Tuple[object, Dict[Compound, str]]:
"""Read bounds from a Pandas DataFrame.
......@@ -261,35 +261,43 @@ class Bounds(BaseBounds):
# legacy code, before we always assumed that the compound IDs are
# from KEGG
if "Compound:Identifiers:kegg.compound" in df.columns:
df["Compound:Identifiers"] = "KEGG:" + df[
"Compound:Identifiers:kegg.compound"]
df["Compound:Identifiers"] = (
"KEGG:" + df["Compound:Identifiers:kegg.compound"]
)
# now, by default, we assume the namespaces are provided as part of the
# reaction formulas
if "Compound:Identifiers" not in df.columns:
raise KeyError("Could not find a column of Compound:Identifiers "
"in the ConcentrationConstraints table")
raise KeyError(
"Could not find a column of Compound:Identifiers "
"in the ConcentrationConstraints table"
)
df["Compound"] = df["Compound:Identifiers"].apply(ccache.get_compound)
if pd.isnull(df.Compound).any():
accessions_not_found = df.loc[pd.isnull(df.Compound),
"Compound:Identifiers"]
accessions_not_found = df.loc[
pd.isnull(df.Compound), "Compound:Identifiers"
]
error_msg = str(accessions_not_found.to_dict())
raise KeyError(f"Some compounds not found in equilibrator-cache: "
f"{error_msg}")
raise KeyError(
f"Some compounds not found in equilibrator-cache: "
f"{error_msg}"
)
name_to_compound = df.Compound.to_dict()
df.set_index("Compound", inplace=True)
if bounds_unit is not None:
lbs = df['Concentration:Min'].apply(lambda x: Q_(float(x),
bounds_unit))
ubs = df['Concentration:Max'].apply(lambda x: Q_(float(x),
bounds_unit))
lbs = df["Concentration:Min"].apply(
lambda x: Q_(float(x), bounds_unit)
)
ubs = df["Concentration:Max"].apply(
lambda x: Q_(float(x), bounds_unit)
)
else:
lbs = df['Concentration:Min'].apply(Q_)
ubs = df['Concentration:Max'].apply(Q_)
lbs = df["Concentration:Min"].apply(Q_)
ubs = df["Concentration:Max"].apply(Q_)
bounds = Bounds(lbs.to_dict(), ubs.to_dict(), default_lb, default_ub)
bounds._check_bounds()
......@@ -314,9 +322,7 @@ class Bounds(BaseBounds):
"""Return a deep copy of self."""
new_lb = dict(self.lower_bounds.items())
new_ub = dict(self.upper_bounds.items())
return Bounds(new_lb, new_ub,
self.default_lb,
self.default_ub)
return Bounds(new_lb, new_ub, self.default_lb, self.default_ub)
def GetLowerBound(self, compound: Compound) -> float:
"""Get the lower bound for this key.
......@@ -347,7 +353,8 @@ class Bounds(BaseBounds):
"""
if Bounds.DEFAULT_BOUNDS is None:
COFACTORS_FNAME = pkg_resources.resource_filename(
'equilibrator_api', 'data/cofactors.csv')
with open(COFACTORS_FNAME, 'r') as fp:
"equilibrator_api", "data/cofactors.csv"
)
with open(COFACTORS_FNAME, "r") as fp:
Bounds.DEFAULT_BOUNDS = Bounds.from_csv(fp)
return Bounds.DEFAULT_BOUNDS
......@@ -43,13 +43,13 @@ class ComponentContribution(object):
predictor = GibbsEnergyPredictor()
@ureg.check(None, None, None, '[concentration]', '[temperature]')
@ureg.check(None, None, None, "[concentration]", "[temperature]")
def __init__(
self,
p_h: float = default_pH,
p_mg: float = default_pMg,
ionic_strength: float = default_I,
temperature: float = default_T
temperature: float = default_T,
) -> object:
"""Create a ComponentContribution object.
......@@ -79,15 +79,16 @@ class ComponentContribution(object):
"""
residual_reaction, stored_dg = reaction.separate_stored_dg()
standard_dg, dg_sigma = \
ComponentContribution.predictor.standard_dg(
residual_reaction)
standard_dg, dg_sigma = ComponentContribution.predictor.standard_dg(
residual_reaction
)
return standard_dg + stored_dg, dg_sigma
@staticmethod
def standard_dg_multi(
reactions: List[PhasedReaction]) -> Tuple[np.array, np.array]:
reactions: List[PhasedReaction]
) -> Tuple[np.array, np.array]:
"""Calculate the chemical reaction energies of a list of reactions.
Using the major microspecies of each of the reactants.
......@@ -99,14 +100,13 @@ class ComponentContribution(object):
residual_reactions, stored_dg = zip(*rxn_dg_pairs)
stored_dg = np.array(stored_dg)
standard_dg, dg_sigma = \
ComponentContribution.predictor.standard_dg_multi(
residual_reactions)
standard_dg, dg_sigma = ComponentContribution.predictor.standard_dg_multi(
residual_reactions
)
return standard_dg + stored_dg, dg_sigma
def standard_dg_prime(
self,
reaction: PhasedReaction
self, reaction: PhasedReaction
) -> Tuple[float, float]:
"""Calculate the transformed reaction energies of a reaction.
......@@ -115,27 +115,22 @@ class ComponentContribution(object):
calculate the confidence interval, use the range -1.96 to 1.96 times
this value
"""
residual_reaction, stored_dg_prime = \
reaction.separate_stored_dg_prime(
p_h=self.p_h,
ionic_strength=self.ionic_strength,
temperature=self.temperature
)
residual_reaction, stored_dg_prime = reaction.separate_stored_dg_prime(
p_h=self.p_h,
ionic_strength=self.ionic_strength,
temperature=self.temperature,
)
standard_dg_prime, dg_sigma = \
ComponentContribution.predictor.standard_dg_prime(
residual_reaction,
p_h=self.p_h,
ionic_strength=self.ionic_strength,
temperature=self.temperature
)
standard_dg_prime, dg_sigma = ComponentContribution.predictor.standard_dg_prime(
residual_reaction,
p_h=self.p_h,
ionic_strength=self.ionic_strength,
temperature=self.temperature,
)
return standard_dg_prime + stored_dg_prime, dg_sigma
def dg_prime(
self,
reaction: PhasedReaction
) -> Tuple[float, float]:
def dg_prime(self, reaction: PhasedReaction) -> Tuple[float, float]:
"""Calculate the dG'0 of a single reaction.
:param reaction: an object of type Reaction
......@@ -149,8 +144,7 @@ class ComponentContribution(object):
return dg_prime, dg_uncertainty
def standard_dg_prime_multi(
self,
reactions: List[PhasedReaction]
self, reactions: List[PhasedReaction]
) -> Tuple[np.array, np.array]:
"""Calculate the transformed reaction energies of a list of reactions.
......@@ -162,24 +156,27 @@ class ComponentContribution(object):
U as the covariance of a Gaussian used for sampling
(where dG0_cc is the mean of that Gaussian).
"""
rxn_dg_pairs = map(lambda r: r.separate_stored_dg_prime(
p_h=self.p_h, ionic_strength=self.ionic_strength,
temperature=self.temperature), reactions)
rxn_dg_pairs = map(
lambda r: r.separate_stored_dg_prime(
p_h=self.p_h,
ionic_strength=self.ionic_strength,
temperature=self.temperature,
),
reactions,
)
residual_reactions, stored_dg_primes = zip(*rxn_dg_pairs)
stored_dg_primes = np.array(stored_dg_primes)
standard_dg_prime, dg_sigma = \
ComponentContribution.predictor.standard_dg_prime_multi(
residual_reactions,
p_h=self.p_h,
ionic_strength=self.ionic_strength,
temperature=self.temperature
)
standard_dg_prime, dg_sigma = ComponentContribution.predictor.standard_dg_prime_multi(
residual_reactions,
p_h=self.p_h,
ionic_strength=self.ionic_strength,
temperature=self.temperature,
)
return standard_dg_prime + stored_dg_primes, dg_sigma
def physiological_dg_prime(
self,
reaction: PhasedReaction
self, reaction: PhasedReaction
) -> Tuple[float, float]:
"""Calculate the dG'm of a single reaction.
......@@ -193,9 +190,11 @@ class ComponentContribution(object):
confidence interval (which is the value shown on eQuilibrator).
"""
physiological_dg_prime, dg_uncertainty = self.standard_dg_prime(
reaction)
physiological_dg_prime += \
reaction
)
physiological_dg_prime += (
self.RT * reaction.physiological_dg_correction()
)
return physiological_dg_prime, dg_uncertainty
def ln_reversibility_index(self, reaction: PhasedReaction) -> float:
......@@ -223,21 +222,21 @@ class ComponentContribution(object):
"""
n_e = reaction.check_half_reaction_balancing()
if n_e is None:
raise ValueError('reaction is not chemically balanced')
raise ValueError("reaction is not chemically balanced")
if n_e == 0:
raise ValueError('this is not a half-reaction, '
'electrons are balanced')
raise ValueError(
"this is not a half-reaction, " "electrons are balanced"
)
dG0_prime, dG0_uncertainty = self.standard_dg_prime(reaction)
E0_prime = -dG0_prime / (n_e*FARADAY) # in Volts
E0_uncertainty = dG0_uncertainty / abs(n_e*FARADAY) # in Volts
E0_prime = -dG0_prime / (n_e * FARADAY) # in Volts
E0_uncertainty = dG0_uncertainty / abs(n_e * FARADAY) # in Volts
return E0_prime, E0_uncertainty
def physiological_e_prime(
self,
reaction: PhasedReaction
self, reaction: PhasedReaction
) -> Tuple[float, float]:
"""Calculate the E'0 of a single half-reaction.
......@@ -250,22 +249,20 @@ class ComponentContribution(object):
"""
n_e = reaction.check_half_reaction_balancing()
if n_e is None:
raise ValueError('reaction is not chemically balanced')
raise ValueError("reaction is not chemically balanced")
if n_e == 0:
raise ValueError('this is not a half-reaction, '
'electrons are balanced')
raise ValueError(
"this is not a half-reaction, " "electrons are balanced"
)
dG0_prime, dG0_uncertainty = self.physiological_dg_prime(reaction)
E0_prime = -dG0_prime / (n_e*FARADAY) # in Volts
E0_uncertainty = dG0_uncertainty / abs(n_e*FARADAY) # in Volts
E0_prime = -dG0_prime / (n_e * FARADAY) # in Volts
E0_uncertainty = dG0_uncertainty / abs(n_e * FARADAY) # in Volts
return E0_prime, E0_uncertainty
def e_prime(
self,
reaction: PhasedReaction
) -> Tuple[float, float]:
def e_prime(self, reaction: PhasedReaction) -> Tuple[float, float]:
"""Calculate the E'0 of a single half-reaction.
:param reaction: a PhasedReaction object
......@@ -277,22 +274,20 @@ class ComponentContribution(object):
"""
n_e = reaction.check_half_reaction_balancing()
if n_e is None:
raise ValueError('reaction is not chemically balanced')
raise ValueError("reaction is not chemically balanced")
if n_e == 0:
raise ValueError('this is not a half-reaction, '
'electrons are balanced')
raise ValueError(
"this is not a half-reaction, " "electrons are balanced"
)
dG0_prime, dG0_uncertainty = self.dg_prime(reaction)
E0_prime = -dG0_prime / (n_e*FARADAY) # in Volts
E0_uncertainty = dG0_uncertainty / abs(n_e*FARADAY) # in Volts
E0_prime = -dG0_prime / (n_e * FARADAY) # in Volts
E0_uncertainty = dG0_uncertainty / abs(n_e * FARADAY) # in Volts
return E0_prime, E0_uncertainty
def dg_analysis(
self,
reaction: PhasedReaction
) -> List[Dict[str, object]]:
def dg_analysis(self, reaction: PhasedReaction) -> List[Dict[str, object]]:
"""Get the analysis of the component contribution estimation process.
:param reaction: a PhasedReaction object.
......
......@@ -34,12 +34,21 @@ from typing import Callable, Iterable, List, TextIO, Tuple
import numpy as np
from equilibrator_cache.reaction import (
create_stoichiometric_matrix_from_reactions)
create_stoichiometric_matrix_from_reactions,
)
from sbtab import SBtab
from scipy.linalg import fractional_matrix_power
from . import (
Q_, Compound, R, default_I, default_pH, default_pMg, default_T, strip_units)
Q_,
Compound,
R,
default_I,
default_pH,
default_pMg,
default_T,
strip_units,
)
from .bounds import Bounds
from .component_contribution import ComponentContribution
from .phased_reaction import PhasedReaction
......@@ -52,17 +61,18 @@ class Pathway(object):
Designed for checking input prior to converting to a stoichiometric model.
"""
def __init__(self,
reactions: List[PhasedReaction],
fluxes: np.array,
standard_dg_primes: np.array = None,
dg_sigma: np.array = None,
bounds: Bounds = None,
p_h: float = default_pH,
p_mg: float = default_pMg,
ionic_strength: float = default_I,
temperature: float = default_T
) -> object:
def __init__(
self,
reactions: List[PhasedReaction],
fluxes: np.array,
standard_dg_primes: np.array = None,
dg_sigma: np.array = None,
bounds: Bounds = None,
p_h: float = default_pH,
p_mg: float = default_pMg,
ionic_strength: float = default_I,
temperature: float = default_T,
) -> object:
"""Initialize.
Args:
......@@ -85,15 +95,20 @@ class Pathway(object):
self.S = create_stoichiometric_matrix_from_reactions(reactions)
self.fluxes = fluxes
assert self.fluxes.shape == (Nr, )
assert self.fluxes.shape == (Nr,)
self.comp_contrib = None
if standard_dg_primes is None:
assert dg_sigma is None, ("If standard_dg_primes are not "
"provided, dg_sigme must also be None")
assert dg_sigma is None, (
"If standard_dg_primes are not "
"provided, dg_sigme must also be None"
)
self.set_aqueous_params(
p_h=p_h, p_mg=p_mg, ionic_strength=ionic_strength,
temperature=temperature)
p_h=p_h,
p_mg=p_mg,
ionic_strength=ionic_strength,
temperature=temperature,
)
else:
assert standard_dg_primes.shape == (Nr,)
# dGr should be orthogonal to nullspace of S
......@@ -126,7 +141,7 @@ class Pathway(object):
p_h: float = None,
p_mg: float = None,
ionic_strength: float = None,
temperature: float = None
temperature: float = None,
) -> None:
"""Set the aqueous conditions and recalculate the standard dG' values.
......@@ -141,8 +156,10 @@ class Pathway(object):
ionic_strength = ionic_strength or default_I
temperature = temperature or default_T
self.comp_contrib = ComponentContribution(
p_h=p_h, p_mg=p_mg, ionic_strength=ionic_strength,
temperature=temperature
p_h=p_h,
p_mg=p_mg,
ionic_strength=ionic_strength,
temperature=temperature,
)
else:
if p_h is not None:
......@@ -156,8 +173,9 @@ class Pathway(object):
RT = R * self.comp_contrib.temperature
standard_dg_primes, dg_uncertainties = \