Source code for eqc_models.allocation.allocation

# (C) Quantum Computing Inc., 2024.
from typing import List, Tuple
from enum import Enum
import numpy as np
from eqc_models.base.constraints import InequalitiesMixin
from eqc_models.base.quadratic import ConstrainedQuadraticModel


[docs] class AllocationModel(ConstrainedQuadraticModel): """ Parameters ---------- resources: List names of available resources. tasks: List names of tasks. resource_usage: List of Lists or 2D np.ndarray rows represent tasks and columns represent resources, specifying the amount of each resource required per task. resource_limits: 1D array or List specifying the limit on each resource. cost_per_task: 1D array List specifying the cost per task (or benefit with negative of value). Attributes ---------- penalty_multiplier: float value for weighting the penalties formed from the equality constraints qubo: eqc_models.base.operators.QUBO QUBO operator representation polynomial: eqc_models.base.operators.Polynomial Polynomial operator representation This class represents a resource allocation model for maximizing total benefit. In other words, Given a list of resources and a list of tasks, allocate the resources among the tasks so as to maximize the economic benefit. Here's an example. Five tasks must share 4 resources. Each task can use a different amount of each resource. +--------------------+------+------+---------+---------+---------+ | | Spam | Eggs | Coconut | Sparrow | Benefit | +--------------------+------+------+---------+---------+---------+ | Breakfast | 1 | 2 | 0 | 0 | 3 | +--------------------+------+------+---------+---------+---------+ | Countryside Stroll | 0 | 0 | 1 | 0 | 1 | +--------------------+------+------+---------+---------+---------+ | Storm Castle | 0 | 12 | 1 | 1 | 10 | +--------------------+------+------+---------+---------+---------+ | Availability | 1 | 12 | 2 | 1 | | +--------------------+------+------+---------+---------+---------+ >>> resources = ["Spam", "Eggs", "Coconut", "Sparrow"] >>> tasks = ["Breakfast", "Countryside Stroll", "Storm Castle"] >>> resource_usage = [[1, 2, 0, 0], [0, 0, 1, 0], [0, 12, 1, 1]] >>> resource_limits = [1, 12, 2, 1] >>> cost_per_task = [-3, -1, -10.] >>> allocation_model = AllocationModel(resources, tasks, resource_usage, resource_limits, cost_per_task) >>> allocation_model.penalty_multiplier = 1 >>> C, J = allocation_model.H >>> C # -3 -2 * (12 * 2 + 1 * 1), -1 -2 * 2*1, -10 -2 * (12 * 12 + 1 * 2 + 1 * 1) ... # doctest: +NORMALIZE_WHITESPACE array([ -53., -5., -304.]) >>> J # doctest: +NORMALIZE_WHITESPACE array([[ 5., 0., 24.], [ 0., 1., 1.], [ 24., 1., 146.]]) """ def __init__(self, resources: List, tasks: List, resource_usage: List, resource_limits: List, cost_per_task: List): if not isinstance(resources, list): raise TypeError("Argument 'resources' must be a list") if not isinstance(tasks, list): raise TypeError("Argument 'tasks' must be a list") if not isinstance(resource_usage, list): raise TypeError("Argument 'resource_usage' must be a list") if not isinstance(resource_limits, list): raise TypeError("Argument 'resource_limits' must be a list") if not isinstance(cost_per_task, list): raise TypeError("Argument 'cost_per_task' must be a list") # PARENT CLASS MUST HAVE TYPE CHECK self.resources = resources self.tasks = tasks self.resource_usage = np.array(resource_usage) self.resource_limits = np.array(resource_limits) self.cost_per_task = np.array(cost_per_task) super(AllocationModel, self).__init__(self.cost_per_task, np.zeros((len(self.tasks), len(self.tasks))), self.resource_usage.T, self.resource_limits) # self.domains = np.array([np.floor(max(self.resource_limits[self.resource_limits != 0]) / # min(self.resource_usage[self.resource_usage != 0]))] * len(self.tasks), dtype=int) # self.upper_bound = [np.floor(max(self.resource_limits[self.resource_limits != 0]) / # min(self.resource_usage[self.resource_usage != 0]))] * len(self.tasks) self._validate_dimensions() ### NEED TO MAKE SURE BASE CLASS VALIDATES self.upper_bound ### @property def upper_bound(self) -> np.array: return np.array([np.floor(max(self.resource_limits[self.resource_limits != 0]) / min(self.resource_usage[self.resource_usage != 0]))] * len(self.tasks), dtype=int) @upper_bound.setter def upper_bound(self, value: List): self._upper_bound = value @property def variables(self): return [(i, j) for i in self.tasks for j in self.resources] def _validate_dimensions(self): """Raises ValueErrors for inconsistent dimensions.""" # Check resource_usage dimensions num_tasks = len(self.tasks) num_resources = len(self.resources) # print("RESOURCES", num_resources) # print("TASKS", num_tasks) # print(self.resource_usage.shape) if self.resource_usage.shape != (num_tasks, num_resources): raise ValueError("resource_usage matrix dimensions don't match number of tasks and resources") # Check resource_limits length if self.resource_limits.shape[0] != num_resources: raise ValueError("resource_limits length doesn't match number of resources") # Check cost_per_task length if self.cost_per_task.shape[0] != num_tasks: raise ValueError("cost_per_task length doesn't match number of tasks")
[docs] def add_task(self, task: str, task_resource_usage: List, task_cost: float): """ task: str Name of task to add task_resource_usage: List Quantity of resource used for task for all tasks task_cost: float Quantity indicating the cost or benefit (negative) of the task Add a task to the problem, modifying the resource usage and task cost arrays. """ self.tasks += [task] self.resource_usage = np.vstack([self.resource_usage, task_resource_usage]) self.cost_per_task = np.append(self.cost_per_task, task_cost)
@property def H(self) -> Tuple[np.ndarray,np.ndarray]: """ Return linear, quadratic portions of the (quadratic) Hamiltonian """ Pl, Pq = self.penalties alpha = self.penalty_multiplier obj_linear, obj_quad = self.linear_objective, self.quad_objective #self._linear_objective(), self._quadratic_objective() self._C = obj_linear + alpha * Pl self._J = obj_quad + alpha * Pq return obj_linear + alpha * Pl, obj_quad + alpha * Pq
[docs] class ResourceRuleEnum(Enum): """ Enumeration of the allowed resource rules, mapping to the mathematical expression: MAXIMUM -> LE (less than or equal to) MINIMUM -> GE (greater than or equal to) EXACT -> EQ (equal to) """ MAXIMUM = "LE" MINIMUM = "GE" EXACT = "EQ"
[docs] class AllocationModelX(InequalitiesMixin, AllocationModel): """ Parameters ---------- resources: List names of available resources. tasks: List names of tasks. resource_usage: List of Lists or 2D np.ndarray rows represent tasks and columns represent resources, specifying the amount of each resource required per task. resource_limits: 1D array or List specifying the limit on each resource. resource_rule: List ResourceRuleEnum values for each resource cost_per_task: 1D array List specifying the cost per task (or benefit with negative of value). Attributes ---------- penalty_multiplier: float value for weighting the penalties formed from the equality constraints qubo: eqc_models.base.operators.QUBO QUBO oeprator representation polynomial: eqc_models.base.operators.Polynomial Polynomial operator representation variables: List names of variables formed from tasks and assignments This class represents a resource allocation model for maximizing total benefit. In other words, Given a list of resources and a list of tasks, allocate the resources among the tasks so as to maximize the economic benefit. Adds resource_rule as an argument. This must be a list of strings specifying constraints for each resource (LE, GE, or EQ). Here's an example. Five tasks must share 4 resources. Each task can use a different amount of each resource. +--------------------+------+------+---------+---------+---------+ | | Spam | Eggs | Coconut | Sparrow | Benefit | +--------------------+------+------+---------+---------+---------+ | Breakfast | 1 | 2 | 0 | 0 | 3 | +--------------------+------+------+---------+---------+---------+ | Countryside Stroll | 0 | 0 | 1 | 0 | 1 | +--------------------+------+------+---------+---------+---------+ | Storm Castle | 0 | 12 | 1 | 1 | 10 | +--------------------+------+------+---------+---------+---------+ | Availability | 1 | 12 | 2 | 1 | | +--------------------+------+------+---------+---------+---------+ >>> resources = ["Spam", "Eggs", "Coconut", "Sparrow"] >>> tasks = ["Breakfast", "Countryside Stroll", "Storm Castle"] >>> resource_usage = [[1, 2, 0, 0], [0, 0, 1, 0], [0, 12, 1, 1]] >>> resource_limits = [1, 12, 2, 1] >>> cost_per_task = [-3, -1, -10.] >>> resource_rules = [ResourceRuleEnum.MAXIMUM for i in range(len(resources))] >>> allocation_model = AllocationModelX(resources, tasks, resource_usage, resource_limits, resource_rules, cost_per_task) >>> allocation_model.penalty_multiplier = 1 >>> C, J = allocation_model.H >>> C # -3 -2 * (12 * 2 + 1 * 1), -1 -2 * 2*1, -10 -2 * (12 * 12 + 1 * 2 + 1 * 1) ... # doctest: +NORMALIZE_WHITESPACE array([ -53., -5., -304., -2., -24., -4., -2.]) >>> J # doctest: +NORMALIZE_WHITESPACE array([[ 5., 0., 24., 1., 2., 0., 0.], [ 0., 1., 1., 0., 0., 1., 0.], [ 24., 1., 146., 0., 12., 1., 1.], [ 1., 0., 0., 1., 0., 0., 0.], [ 2., 0., 12., 0., 1., 0., 0.], [ 0., 1., 1., 0., 0., 1., 0.], [ 0., 0., 1., 0., 0., 0., 1.]]) """ def __init__(self, resources: List, tasks: List, resource_usage: List, resource_limits: List, resource_rule: List[ResourceRuleEnum], cost_per_task: List): super().__init__(resources, tasks, resource_usage, resource_limits, cost_per_task) if not isinstance(resource_rule, list): raise TypeError("Argument 'resource_rule' must be a list") elif len(resource_rule) != len(resources): raise ValueError("Argument 'resource_rule' must be the same length as 'resources'") try: check_rule = set([rule.value for rule in resource_rule]) if not check_rule.issubset({"LE", "GE", "EQ"}): raise ValueError("Argument 'resource_rule' must contain only enums 'ResourceRuleEnum.MAXIMUM', " "'ResourceRuleEnum.MINIMUM' or 'ResourceRuleEnum.EXACT'.") except AttributeError as e: # Handle the case where elements in resource_rule don't have a 'value' attribute (likely not enums) raise TypeError("Argument 'resource_rule' must contain only enums. Elements lack a 'value' " "attribute.") from e self.senses = list([rule.value for rule in resource_rule]) @property def upper_bound(self) -> np.array: return np.array([np.floor(max(self.resource_limits[self.resource_limits != 0]) / min(self.resource_usage[self.resource_usage != 0]))] * self.n, dtype=int) @upper_bound.setter def upper_bound(self, value: List): self._upper_bound = value @property def linear_objective(self) -> np.ndarray: """ Returns a 1D numpy array representing the linear part of the objective function (total profit). """ return np.hstack([self.cost_per_task, np.zeros(self.num_slacks)]) @linear_objective.setter def linear_objective(self, value : np.ndarray): assert (value[len(self.cost_per_task):]==0).all(), "additional values beyond cost length must be 0" self.cost_per_task[:] = value[:len(self.cost_per_task)] @property def quad_objective(self) -> np.ndarray: """ Returns a 2D numpy array representing the quadratic part of the objective function (always zero in this case). """ # No quadratic term n = self.n return np.zeros((n, n)) @quad_objective.setter def quad_objective(self, value: np.ndarray): """ Don't let anything be passed in except arrays of all 0 """ # the setting of this gets ignored, but speak if somebody brings a soul # that hasn't passed yet (all must be 0) assert (value==0).all(), "quadratic terms in objective must be 0" @property def H(self): """ Overrides the parent build method to incorporate slack variables based on resource_rule. """ # Build constraint penalties with slack variables Pl, Pq = self.penalties alpha = self.penalty_multiplier obj_linear, obj_quad = self.linear_objective, self.quad_objective self._C = obj_linear + alpha * Pl self._J = obj_quad + alpha * Pq return obj_linear + alpha * Pl, obj_quad + alpha * Pq
[docs] def checkResources(self, solution): """ Parameters ---------- solution: List or np.ndarray solution vector to check for resource violations Returns ------- List of violations: (name, rule, violation quantity, message) """ violations = [] solution = np.array(solution) for name, usage, rule, limit in zip(self.resources, self.resource_usage.T, self.senses, self.resource_limits): decision_vars = self.n - self.num_slacks value = np.squeeze(usage@solution[:decision_vars]) if rule == ResourceRuleEnum.MINIMUM: mult = -1 else: mult = 1 if rule == ResourceRuleEnum.EXACT: if value - limit != 0: msg = f"{value} of {name} violates {rule} {limit}" violations.append((name, rule, value - limit, msg)) else: if mult * (value - limit) > 0: msg = f"{value} of {name} violates {rule} {limit}" violations.append((name, rule, (value - limit), msg)) return violations
@property def n(self): return len(self.tasks) + self.num_slacks