# The MIT License (MIT)
#
# Copyright (c) 2026 Department of Plant and Environmental Science,
# Weizmann Institute of Science.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import datetime
import os
import tarfile
import time
from typing import Iterator, List
from warnings import warn
import cvxpy as cp
import pandas as pd
from . import ALL_VARIABLES
from .io import JSON_EXAMPLE_URL, read_arguments_from_url
from .model_balancing_cvx import ModelBalancingConvex, NegativeFluxError
from .model_balancing_noncvx import ModelBalancing
[docs]
def get_convex_solution(args: dict) -> dict:
try:
mbc = ModelBalancingConvex(**args)
except NegativeFluxError:
warn(
"One of the fluxes is negative, this is not yet supported in "
"the convex version of model balancing"
)
return None
mbc.initialize_with_gmeans()
if not mbc.is_gmean_feasible():
if mbc.find_inner_point(verbose=False) not in cp.settings.SOLUTION_PRESENT:
warn("Cannot find an inner point given the constraints")
return None
if mbc.solve(verbose=False) in cp.settings.SOLUTION_PRESENT:
return {
f"ln_{p}": mbc._var_dict[f"ln_{p}"].value
for p in ALL_VARIABLES
if mbc._var_dict[f"ln_{p}"] is not None
}
[docs]
def run_batch(
example_names: Iterator[str],
init_methods: List[str],
alphas: List[float],
basinhopping_kwargs: dict,
minimizer_kwargs: dict,
) -> None:
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
z_scores_data = []
with tarfile.open(
f"res/model_balancing_{timestamp}.tar.gz", "w:gz"
) as result_tarfile:
for example_name in example_names:
print(f"Analyzing example: {example_name}")
args = read_arguments_from_url(JSON_EXAMPLE_URL + example_name + ".json")
for init_method in init_methods:
for a in alphas:
result_dict = {
"JSON": example_name,
"alpha": a,
"beta": 0.0,
}
args["alpha"] = a
args["beta"] = 0.0
mb = ModelBalancing(**args)
if init_method == "convex":
initial_point = get_convex_solution(args)
if initial_point is None:
continue
mb._var_dict.update(initial_point)
result_dict["initialization"] = "convex solution"
elif init_method == "geom_mean":
mb._var_dict.update(mb.ln_geom_mean)
result_dict["initialization"] = "geometric means"
elif init_method == "true_value":
mb._var_dict.update(mb.ln_true_value)
result_dict["initialization"] = "true values"
print(
f"Initializing non-convex solver with {result_dict['initialization']}, "
f"α = {a:5.1g} ... ",
end="",
)
result_dict["objective_before_solving"] = mb.objective_value
tic = time.perf_counter()
result = mb.solve(
basinhopping_kwargs=basinhopping_kwargs,
minimizer_kwargs=minimizer_kwargs,
)
toc = time.perf_counter()
result_dict["runtime"] = toc - tic
result_dict["success"] = result.success
result_dict["message"] = result.message
print(
f"solver message = {result.message}, "
f"optimized total squared Z-scores = {mb.objective_value:.3f}"
)
result_dict.update(mb.get_z_scores())
result_dict["objective_after_solving"] = mb.objective_value
z_scores_data.append(result_dict)
state_fname = (
f"/tmp/{example_name}_a{a:.1g}_{init_method}_state.tsv"
)
model_fname = (
f"/tmp/{example_name}_a{a:.1g}_{init_method}_model.tsv"
)
with open(state_fname, "wt") as fp:
fp.write(mb.to_state_sbtab().to_str())
result_tarfile.add(state_fname)
os.remove(state_fname)
with open(model_fname, "wt") as fp:
fp.write(mb.to_model_sbtab().to_str())
result_tarfile.add(model_fname)
os.remove(model_fname)
df = pd.DataFrame.from_dict(z_scores_data).set_index(["JSON", "alpha"])
summary_fname = "/tmp/summary.csv"
df.round(5).to_csv(summary_fname)
result_tarfile.add(summary_fname)