# Source code for picos.reforms.reformulation

```
# ------------------------------------------------------------------------------
# Copyright (C) 2019 Maximilian Stahlberg
#
# This file is part of PICOS.
#
# PICOS is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# PICOS is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# ------------------------------------------------------------------------------
"""Backend for problem reformulation classes."""
# ------------------------------------------------------------------------------
# TODO: Make the solver base class inherit from Reformulation, in particular the
# update related methods.
# ------------------------------------------------------------------------------
from __future__ import print_function
from ..apidoc import api_end, api_start
from ..compat import ABC, abstractmethod
from ..modeling import Footprint, Problem, Solution
_API_START = api_start(globals())
# -------------------------------
[docs]class Reformulation(ABC):
"""Base class for problem reformulations.
Abstract base class for a reformulation from one (possibly already
reformulated) problem form to another.
"""
# --------------------------------------------------------------------------
# The following section contains the abstract methods, which are exactly the
# methods that reformulation implementations are supposed to override.
# --------------------------------------------------------------------------
[docs] @classmethod
@abstractmethod
def supports(cls, footprint):
"""Whether the reformulation affects problems with the given footprint.
The reformulation must support every problem with such a footprint and
the resulting problem should have a changed footprint.
"""
pass
[docs] @classmethod
@abstractmethod
def predict(cls, footprint):
"""Predict the reformulation's effect on a problem footprint.
Given a problem footprint, returns another problem footprint that a
problem with the former one would be reformulated to.
This is used to predict the effects of a reformulation when planning
a solution strategy without the cost of actually transforming a problem.
"""
pass
[docs] @abstractmethod
def forward(self):
"""Perform the initial problem reformulation.
Creates a modified copy or clone of the problem in :attr:`input` and
stores it as :attr:`output`.
See :func:`copy <picos.Problem.copy>` and :func:`clone
<picos.Problem.clone>` for the differences between a copy and a clone.
Implementations are supposed to do the necessary bookkeeping so that
:func:`backward` can transform a solution to the new problem back to a
solution of the original problem.
"""
pass
[docs] @abstractmethod
def update(self):
"""Update a previous problem reformulation.
Updates :attr:`output` and related bookkeeping information with respect
to changes in :attr:`input`.
:raises NotImplementedError:
If performing an update is not feasible for the reformulation.
"""
pass
[docs] @abstractmethod
def backward(self, solution):
"""Translate back a solution from reformulated to original problem.
Transforms a single :class:`solution <picos.Solution>` to :attr:`output`
to a solution of :attr:`input`.
The method is allowed to modify the solution; it is not necessary to
work on a copy. In particular, :func:`attach_to
<picos.Solution.attach_to>` can be used if :func:`forward` has created
a deep copy of the problem.
"""
pass
# --------------------------------------------------------------------------
# The following section contains the non-abstract methods, which are exactly
# the methods that reformulation implementations are supposed to inherit.
# --------------------------------------------------------------------------
def _reset_knowns(self):
self._knownObjective = None
self._knownVariables = set()
self._knownConstraints = set()
def _set_knowns(self):
self._knownObjective = self.input.objective
self._knownVariables = set(self.input.variables.values())
self._knownConstraints = set(self.input.constraints.values())
[docs] def __init__(self, theObject):
"""Initialize :class:`Reformulation` instances.
:param theObject: The input to work on; either an optimization problem
or the (future) output of another reformulation.
:type theObject:
~picos.Problem or ~picos.reforms.reformulation.Reformulation
"""
if isinstance(theObject, Problem):
self.predecessor = None
self._input = theObject
elif isinstance(theObject, Reformulation):
self.predecessor = theObject
theObject.successor = self
elif isinstance(theObject, Footprint):
raise TypeError("Reformulations cannot be instanciated using "
"problem footprints. Use the predict classmethod.")
else:
raise TypeError("Cannot reformulate an object of type '{}'."
.format(type(theObject).__name__))
self.successor = None
"""The next reformulation in the pipeline."""
self.output = None
"""The output problem."""
self._reset_knowns()
[docs] def reset(self):
"""Reset the pipeline from this reformulation onward.
This is done whenever a reformulation does not implement :meth:`update`
so that succeeding reformulations do not attempt to update a problem
which was completely rewritten as this may be inefficient.
"""
assert self.successor, \
"The reformulation being reset has no successor."
self.output = None
self._reset_knowns()
self.successor.reset()
input = property(lambda self:
self.predecessor.output if self.predecessor else self._input,
doc="The input problem.")
verbosity = property(lambda self: self.input.options.verbosity,
doc="Verbosity level of the reformulation; same as for input problem.")
def _verify_prediction(self):
if not self.input.options.verify_prediction:
return
expectedType = self.predict(Footprint.from_problem(self.input))
outputType = Footprint.from_problem(self.output)
if outputType != expectedType:
raise RuntimeError("{} failed to produce a problem with the "
"expected footprint:\nEXPECTED: {}\nOUTCOME : {}\n"
"This is a bug; please report it to the PICOS developers. "
"You can disable the 'verify_prediction' option to try solving "
"anyway.".format(type(self).__name__, expectedType, outputType))
[docs] def execute(self):
"""Reformulate the problem and obtain a solution from the result.
For this to work there needs to be a solver instance at the end of the
reformulation pipeline, which would implement its own version of this
method that actually solves the problem and produces the first solution.
"""
assert self.successor, \
"The reformulation being executed has no successor."
verbose = self.verbosity > 0
# Update the output problem if possible.
if self.output:
if verbose:
print("Updating {}.".format(self.__class__.__name__))
try:
self.update()
except NotImplementedError:
if verbose:
print("Update failed: Not implemented for {}."
.format(self.__class__.__name__))
self.reset()
# Create the output problem if necessary.
if not self.output:
if verbose:
print("Applying {}.".format(self.__class__.__name__))
self.forward()
self._set_knowns()
# Verify that the output problem is of the expected type.
self._verify_prediction()
# Advance one step.
outputSolution = self.successor.execute()
# Transform the solution of the output problem for the input problem.
if isinstance(outputSolution, Solution):
return self.backward(outputSolution)
else:
return [self.backward(solution) for solution in outputSolution]
def _objective_has_changed(self):
"""Check for an objective function change.
:returns: Whether the optimization objective has changed since the last
forward or update.
"""
assert self._knownObjective is not None, \
"_objective_has_changed may only be used inside _update_problem."
objectiveChanged = self._knownObjective != self.input.objective
if objectiveChanged:
self._knownObjective = self.input.objective
return objectiveChanged
def _new_variables(self):
"""Check for new variables.
Yields variables that were added to the input problem since the last
forward or update.
Note that variables received from this method will also be added to the
set of known variables, so you can only iterate once within each update.
"""
for variable in self.input.variables.values():
if variable not in self._knownVariables:
self._knownVariables.add(variable)
yield variable
def _removed_variables(self):
"""Check for removed variables.
Yields variables that were removed from the input problem since the last
forward or update.
Note that variables received from this method will also be removed from
the set of known variables, so you can only iterate once within each
update.
"""
newVariables = set(self.input.variables.values())
for variable in self._knownVariables:
if variable not in newVariables:
yield variable
self._knownVariables.intersection_update(newVariables)
def _new_constraints(self):
"""Check for new constraints.
Yields constraints that were added to the input problem since the last
forward or update.
Note that constraints received from this method will also be added to
the set of known constraints, so you can only iterate once within each
update.
"""
for constraint in self.input.constraints.values():
if constraint not in self._knownConstraints:
self._knownConstraints.add(constraint)
yield constraint
def _removed_constraints(self):
"""Check for removed constraints.
Yields constraints that were removed from the input problem since the
last forward or update.
Note that constraints received from this method will also be removed
from the set of known constraints, so you can only iterate once within
each update.
"""
newConstraints = set(self.input.constraints.values())
for constraint in self._knownConstraints:
if constraint not in newConstraints:
yield constraint
self._knownConstraints.intersection_update(newConstraints)
def _pass_updated_objective(self):
"""Pass changes in the objective function from input to output problem.
.. warning::
This method resets the objective-has-changed state.
"""
if self._objective_has_changed():
self.output.objective = self.input.objective
# TODO: Determine if and in which form such a method is needed now that
# variables are added to problems implicitly (but still explicitly to
# solvers).
def _pass_updated_vars(self):
"""Pass variable changes from input to output problem.
Adds all new varibles in the input problem to the output problem, and
removes all variables removed from the input problem also from the
output problem.
.. warning::
Variables are passed as with :meth:`Problem.clone`, not copied.
.. warning::
This method clears the buffers of new and removed variables.
"""
for variable in self._new_variables():
pass
for variable in self._removed_variables():
pass
def _pass_updated_cons(self, ignore=type(None)):
"""Pass constraint changes from input to output problem.
Adds all new constraints in the input problem to the output problem, and
removes all constraints removed from the input problem also from the
output problem.
:param type ignore: Constraints of this type are not handled. Instead,
the method returns a pair `(added, removed)` that contains the
respective constraints that were not handled.
.. warning::
Constraints are passed as with :meth:`Problem.clone`, not copied.
.. warning::
This method clears the buffers of new and removed constraints.
"""
added, removed = [], []
for constraint in self._new_constraints():
assert constraint.id not in self.output.constraints
if isinstance(constraint, ignore):
added.append(constraint)
else:
self.output.add_constraint(constraint)
for constraint in self._removed_constraints():
if isinstance(constraint, ignore):
# No assertion in this case, because the reformulation would not
# have added such a constraint.
removed.append(constraint)
else:
assert constraint.id in self.output.constraints
self.output.remove_constraint(constraint.id)
if ignore is not type(None): # noqa: E721
return added, removed
def _pass_updated_options(self):
"""Make the output problem use the same options object as the input."""
self.output.options = self.input.options
# --------------------------------------
__all__ = api_end(_API_START, globals())
```