8000 fix(pulp): avoid PuLP 4.0-deprecated APIs by devin-ai-integration[bot] · Pull Request #103 · dominicusin/package-maximizer · GitHub
[go: up one dir, main page]

Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
137 changes: 91 additions & 46 deletions package_maximizer/solvers/pulp_solver.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from __future__ import annotations

import logging
import warnings
from typing import TYPE_CHECKING, Iterable, Optional

from ..core.interfaces import ConstraintSolver
Expand All @@ -22,13 +23,44 @@

try:
import pulp
from pulp import LpBinary, LpMaximize, LpProblem, LpStatus, LpVariable, lpSum, value
from pulp import LpBinary, LpMaximize, LpProblem, LpVariable, lpSum, value

PULP_AVAILABLE = True
except ImportError:
PULP_AVAILABLE = False


_SOLVER_ALIASES = {"CBC": "COIN_CMD", "HIGHS": "HiGHS"}


def _binary_vars(prob, prefix: str, names: list[str]) -> dict:
"""Create binary variables via the PuLP 4.0 API when available."""
if hasattr(prob, "add_variable_dicts"):
return prob.add_variable_dicts(prefix, names, cat=LpBinary)
return LpVariable.dicts(prefix, names, cat=LpBinary)


def _bundled_cbc(time_limit: float):
"""PuLP's bundled CBC (``PULP_CBC_CMD``, PuLP < 4.0 only), or ``None``."""
bundled = getattr(pulp, "PULP_CBC_CMD", None)
if bundled is None:
return None
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
return bundled(msg=False, timeLimit=time_limit)


def _status_name(prob, result) -> str:
"""Solve status name for PuLP 4 (``LpSolveStats``) and older (int status)."""
status = getattr(result, "status", None)
if status is None:
< 8000 /td> status = getattr(prob, "status", result)
name = getattr(status, "name", None)
if isinstance(name, str):
return name
return getattr(pulp, "LpStatus", {}).get(status, str(status))


class PulPSolver(ConstraintSolver):
"""ILP/MIP-солвер на основе PuLP."""

Expand All @@ -49,35 +81,7 @@ def solve(self, packages: Iterable[Package]) -> list[str]:
if not package_list:
return []

constraints = encode_packages(package_list)
prob = LpProblem("Package_Maximization", LpMaximize)

# Бинарные переменные
x = LpVariable.dicts("Package", constraints.packages, cat=LpBinary)

# Цель: максимизировать количество
prob += lpSum(x[name] for name in constraints.packages), "Total_Packages"

# Конфликты
for a, b in constraints.conflicts:
prob += x[a] + x[b] <= 1, f"Conflict_{a}_{b}"

# Зависимости: selected(pkg) => selected(dep)
for pkg, deps in constraints.dependencies.items():
for dep in deps:
prob += x[pkg] - x[dep] <= 0, f"Dep_{pkg}_{dep}"

try:
prob.solve(pulp.PULP_CBC_CMD(msg=False, timeLimit=self.time_limit / 1000))
except Exception as e:
logger.warning(f"PuLP solver failed: {e}")
return self._greedy_fallback(package_list)

if LpStatus[prob.status] == "Optimal":
return [name for name in constraints.packages if value(x[name]) == 1]
else:
logger.warning(f"PuLP solver status: {prob.status}")
return self._greedy_fallback(package_list)
return self._solve_ilp(package_list, None, "Package_Maximization")

def solve_with_weights(
self, packages: Iterable[Package], weights: dict[str, float] | None = None
Expand All @@ -95,37 +99,78 @@ def solve_with_weights(
if weights is None:
weights = {pkg.name: 1.0 for pkg in package_list}
8000
constraints = encode_packages(package_list)
prob = LpProblem("Weighted_Package_Maximization", LpMaximize)
return self._solve_ilp(package_list, weights, "Weighted_Package_Maximization")

x = LpVariable.dicts("Package", constraints.packages, cat=LpBinary)
def _solve_ilp(
self,
package_list: list[Package],
weights: dict[str, float] | None,
problem_name: str,
) -> list[str]:
constraints = encode_packages(package_list)
prob = LpProblem(problem_name, LpMaximize)
x = _binary_vars(prob, "Package", constraints.packages)

# Цель: максимизировать вес
prob += (
lpSum(x[name] * weights.get(name, 1.0) for name in constraints.packages),
"Total_Weight",
)
if weights is None:
prob += lpSum(x[name] for name in constraints.packages), "Total_Packages"
else:
prob += (
lpSum(
x[name] * weights.get(name, 1.0) for name in constraints.packages
),
"Total_Weight",
)

# Конфликты
for a, b in constraints.conflicts:
prob += x[a] + x[b] <= 1, f"Conflict_{a}_{b}"

# Зависимости
# selected(pkg) => selected(dep)
for pkg, deps in constraints.dependencies.items():
for dep in deps:
prob += x[pkg] - x[dep] <= 0, f"Dep_{pkg}_{dep}"

try:
prob.solve(pulp.PULP_CBC_CMD(msg=False, timeLimit=self.time_limit / 1000))
result = prob.solve(self._make_backend())
except Exception as e:
logger.warning(f"PuLP solver failed: {e}")
return self._greedy_fallback(package_list, weights)

if LpStatus[prob.status] == "Optimal":
return [name for name in constraints.packages if value(x[name]) == 1]
else:
logger.warning(f"PuLP solver status: {prob.status}")
return self._greedy_fallback(package_list, weights)
status = _status_name(prob, result)
if status == "Optimal":
return [
name for name in constraints.packages if (value(x[name]) or 0) > 0.5
]
logger.warning(f"PuLP solver status: {status}")
return self._greedy_fallback(package_list, weights)

def _make_backend(self):
"""Pick a PuLP backend without relying on APIs removed in PuLP 4.0.

Order: explicit ``solver_name`` → ``COIN_CMD`` (system or ``pulp[cbc]``
CBC) → HiGHS → bundled ``PULP_CBC_CMD`` (PuLP < 4.0 only).
"""
time_limit = self.time_limit / 1000
if self.solver_name:
name = _SOLVER_ALIASES.get(self.solver_name.upper(), self.solver_name)
backend = pulp.getSolver(name, msg=False, timeLimit=time_limit)
if name == "COIN_CMD" and not backend.available():
return _bundled_cbc(time_limit) or backend
return backend

for name in ("COIN_CMD", "HiGHS", "HiGHS_CMD"):
solver_cls = getattr(pulp, name, None)
if solver_cls is None:
continue
candidate = solver_cls(msg=False, timeLimit=time_limit)
if candidate.available():
return candidate

bundled = _bundled_cbc(time_limit)
if bundled is None:
raise RuntimeError(
"No PuLP backend available; install CBC with `pip install pulp[cbc]`"
)
return bundled

def _greedy_fallback(
self, packages: list[Package], weights: dict[str, float] | None = None
Expand Down
130 changes: 130 additions & 0 deletions tests/test_pulp_solver.py
Original file line number Diff line number Diff line change
Expand Up @@ -276,3 +276,133 @@ def test_greedy_fallback_with_weights_produces_valid_result(self):
]
result = solver._greedy_fallback(pkgs, {"a": 1.0, "b": 2.0})
assert not ("a" in result and "b" in result)


class TestPulPBackendSelection:
"""_make_backend avoids PuLP 4.0-deprecated APIs where possible."""

def test_explicit_solver_name_alias(self):
pytest.importorskip("pulp")
from package_maximizer.solvers.pulp_solver import PulPSolver

with patch("package_maximizer.solvers.pulp_solver.pulp.getSolver") as gs:
PulPSolver(solver_name="cbc", time_limit=2000)._make_backend()
gs.assert_called_once_with("COIN_CMD", msg=False, timeLimit=2.0)

def test_explicit_cbc_falls_back_to_bundled_when_coin_cmd_missing(self):
pulp = pytest.importorskip("pulp")
if not hasattr(pulp, "PULP_CBC_CMD"):
pytest.skip("PuLP >= 4.0 has no bundled CBC")
from package_maximizer.solvers.pulp_solver import PulPSolver

with patch.object(pulp.COIN_CMD, "available", return_value=False):
backend = PulPSolver(solver_name="CBC")._make_backend()
assert isinstance(backend, pulp.PULP_CBC_CMD)

def test_explicit_cbc_end_to_end_without_system_cbc(self):
pulp = pytest.importorskip("pulp")
from package_maximizer.solvers.pulp_solver import PulPSolver, _bundled_cbc

bundled = _bundled_cbc(1.0)
if bundled is None or not bundled.available():
pytest.skip("bundled CBC not available")

packages = [
Package(name="A", version="1.0", conflicts=["B", "C"]),
Package(name="B", version="1.0"),
Package(name="C", version="1.0"),
]
solver = PulPSolver(solver_name="CBC")
with (
patch.object(pulp.COIN_CMD, "available", return_value=False),
patch.object(solver, "_greedy_fallback", side_effect=AssertionError),
):
assert sorted(solver.solve(packages)) == ["B", "C"]

def test_explicit_solver_used_when_available(self):
pytest.importorskip("pulp")
from package_maximizer.solvers.pulp_solver import PulPSolver

with patch("package_maximizer.solvers.pulp_solver.pulp.getSolver") as gs:
gs.return_value.available.return_value = False
backend = PulPSolver(solver_name="HiGHS")._make_backend()
assert backend is gs.return_value

def test_prefers_available_coin_cmd(self):
pulp = pytest.importorskip("pulp")
from package_maximizer.solvers.pulp_solver import PulPSolver

with patch.object(pulp.COIN_CMD, "available", return_value="/usr/bin/cbc"):
backend = PulPSolver()._make_backend()
assert type(backend) is pulp.COIN_CMD

def test_bundled_cbc_fallback_without_deprecation_warning(self):
import warnings

pulp = pytest.importorskip("pulp")
if not hasattr(pulp, "PULP_CBC_CMD"):
pytest.skip("PuLP >= 4.0 has no bundled CBC")
from package_maximizer.solvers.pulp_solver import PulPSolver

with (
patch.object(pulp.COIN_CMD, "available", return_value=False),
patch.object(pulp.HiGHS, "available", return_value=False),
patch.object(pulp.HiGHS_CMD, "available", return_value=False),
):
with warnings.catch_warnings():
warnings.simplefilter("error", DeprecationWarning)
backend = PulPSolver()._make_backend()
assert isinstance(backend, pulp.PULP_CBC_CMD)

def test_no_backend_raises_and_solve_falls_back(self):
pulp = pytest.importorskip("pulp")
from package_maximizer.solvers import pulp_solver
from package_maximizer.solvers.pulp_solver import PulPSolver

fake = MagicMock(wraps=pulp)
fake.PULP_CBC_CMD = None
for name in ("COIN_CMD", "HiGHS", "HiGHS_CMD"):
setattr(fake, name, None)
with patch.object(pulp_solver, "pulp", fake):
with pytest.raises(RuntimeError):
PulPSolver()._make_backend()
pkgs = [Package(name="a", conflicts=["b"]), Package(name="b")]
result = PulPSolver().solve(pkgs)
assert result and not ("a" in result and "b" in result)

def test_binary_vars_legacy_api(self):
pulp = pytest.importorskip("pulp")
from package_maximizer.solvers.pulp_solver import _binary_vars

89AD legacy_prob = MagicMock(spec=[])
with patch.object(
pulp.LpVariable, "dicts", return_value={"a": 1}, create=True
) as d:
assert _binary_vars(legacy_prob, "Package", ["a"]) == {"a": 1}
d.assert_called_once()


class TestPulPStatusName:
"""_status_name supports PuLP 4 LpSolveStats and legacy int statuses."""

def test_pulp4_stats_object(self):
pytest.importorskip("pulp")
from package_maximizer.solvers.pulp_solver import _status_name

stats = MagicMock()
stats.status.name = "Optimal"
assert _status_name(MagicMock(spec=[]), stats) == "Optimal"

def test_legacy_int_status(self):
pytest.importorskip("pulp")
from package_maximizer.solvers import pulp_solver
from package_maximizer.solvers.pulp_solver import _status_name

prob = MagicMock(spec=["status"])
prob.status = 1
fake = MagicMock(wraps=pulp_solver.pulp)
fake.LpStatus = {1: "Optimal", 0: "Not Solved"}
with patch.object(pulp_solver, "pulp", fake):
assert _status_name(prob, 1) == "Optimal"
prob.status = 0
assert _status_name(prob, 0) == "Not Solved"
Loading
0