Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_flux_variability_sequential_remove_cycles(self, core_model):
original_objective = core_model.objective
fva_solution = flux_variability_analysis(core_model, fraction_of_optimum=0.999999419892,
remove_cycles=True,
view=SequentialView())
assert REFERENCE_FVA_SOLUTION_ECOLI_CORE['upper_bound']['FRD7'] > 666.
assert round(abs(fva_solution['upper_bound']['FRD7'] - 0.), 7) == 0
for key in fva_solution.data_frame.index:
if REFERENCE_FVA_SOLUTION_ECOLI_CORE['lower_bound'][key] > -666:
assert abs(
fva_solution['lower_bound'][key] - REFERENCE_FVA_SOLUTION_ECOLI_CORE['lower_bound'][key]) < 0.0001
if REFERENCE_FVA_SOLUTION_ECOLI_CORE['upper_bound'][key] < 666:
assert abs(
fva_solution['upper_bound'][key] - REFERENCE_FVA_SOLUTION_ECOLI_CORE['upper_bound'][key]) < 0.0001
cycle_reac = Reaction("minus_PGI") # Create fake cycle
cycle_reac.lower_bound = -1000
core_model.add_reaction(cycle_reac)
cycle_reac.add_metabolites({met: -c for met, c in core_model.reactions.PGI.metabolites.items()})
fva_solution = flux_variability_analysis(core_model, remove_cycles=False, reactions=["PGI"])
assert fva_solution.data_frame.loc["PGI", "upper_bound"] == 1000
def test_one_variable_sequential_metabolite(self, core_model):
ppp = phenotypic_phase_plane(core_model, ['EX_o2_LPAREN_e_RPAREN_'], core_model.metabolites.ac_c,
view=SequentialView())
assert_data_frames_equal(ppp, REFERENCE_PPP_o2_EcoliCore_ac, sort_by=['EX_o2_LPAREN_e_RPAREN_'])
def test_run_single_objective(self, reaction_ko_single_objective):
# TODO: make optlang deterministic so this results can be permanently stored.
_, result_file = mkstemp('.pkl')
results = reaction_ko_single_objective.run(max_evaluations=3000, pop_size=10, view=SequentialView(), seed=SEED)
assert len(results.data_frame.targets) > 0
assert len(results.data_frame.targets) == len(results.data_frame.targets.apply(tuple).unique())
with open(result_file, 'wb') as in_file:
pickle.dump(results, in_file)
with open(result_file, 'rb') as in_file:
if six.PY3:
expected_results = pickle.load(in_file, encoding="latin1")
else:
expected_results = pickle.load(in_file)
assert results.seed == expected_results.seed
def test_run_gene_ko_single_objective_benchmark(self, gene_ko_single_objective, benchmark):
benchmark(gene_ko_single_objective.run, max_evaluations=3000, pop_size=10, view=SequentialView(), seed=SEED)
@property
def effective_lower_bound(self):
model = self.get_model()
return \
flux_variability_analysis(model, reactions=[self], view=SequentialView(), remove_cycles=False)[
'lower_bound'][
self.id]
def run(self, view=config.default_view, **run_kwargs):
run_kwargs['view'] = parallel.SequentialView()
runner = MultiprocessRunner(self._island_class, self._init_kwargs(), self.migrator, run_kwargs)
clients = [[o.clients[i] for o in self.observers] for i in xrange(len(view))]
results = view.map(runner, clients)
return results
# Determine if bokeh is available
# TODO: This should also check if a bokeh server is actually running.
try:
from bokeh.plotting import output_notebook
if in_ipnb():
output_notebook(hide_banner=True)
use_bokeh = True
except ImportError:
use_bokeh = False
bokeh_url = 'default'
# Set default parallelization view
default_view = SequentialView()
def run(self, view=config.default_view, number_of_islands=None, **run_kwargs):
if number_of_islands is None:
number_of_islands = len(view)
run_kwargs['view'] = parallel.SequentialView()
runner = MultiprocessRunner(self._island_class, self._init_kwargs(), self.migrator, run_kwargs)
clients = [[o.clients[i] for o in self.observers] for i in range(number_of_islands)]
try:
results = view.map(runner, clients)
except KeyboardInterrupt as e:
view.shutdown()
raise e
return results
def design(product, host, output, format, cores, aerobic, differential_fva, heuristic_optimization,
max_pathway_predictions, differential_fva_points, pathway_prediction_timeout,
heuristic_optimization_timeout, verbose, logging):
"""Compute strain designs for desired products and host organisms."""
from cameo.api.designer import design
from cameo.api.hosts import hosts
from cameo.parallel import MultiprocessingView, SequentialView
logger.setLevel(logging)
hosts = [getattr(hosts, host) for host in host]
if cores > 1:
view = MultiprocessingView(processes=cores)
elif cores == 1:
view = SequentialView()
design.options.pathway_prediction_timeout = pathway_prediction_timeout * 60
design.options.heuristic_optimization_timeout = heuristic_optimization_timeout
design.options.max_pathway_predictions = max_pathway_predictions
design.options.differential_fva = differential_fva
design.options.heuristic_optimization = heuristic_optimization
design.options.differential_fva_points = differential_fva_points
results = design(product=product, hosts=hosts,
view=view, aerobic=aerobic)
click.echo(results)
for output, format in zip(output, format):
if format == 'pickle':
click.echo('Pickling results into {}'.format(output))
with open(output, 'wb') as f: