kaiwu.hybrid package#

Module contents#

Module: solver

Function: provides a series of solvers based on optimizers in cim/classical

class kaiwu.hybrid.PenaltyMethodOptimizer(optimizer, controller)[source]#

Bases: QuboSolver, JsonSerializableMixin

Penalty function method

Setting kw.common.CheckpointManager.save_dir = ‘/tmp’ enables caching. The current state and penalty coefficient combinations that satisfy hard constraints are cached. After caching is enabled, PenaltyMethodOptimizer can continue iterative solving from where the previous program run terminated. Feasible solutions produced during execution are saved under results in the corresponding JSON file.

Args:

optimizer (Optimizer): optimizer

controller (SolverLoopController): loop controller

Examples:
>>> import kaiwu as kw
>>> import numpy as np
>>> taskNums = 20
>>> machineNums = 5
>>> duration = np.array(range(1, taskNums + 1))
>>> machine_start_time = np.array(range(1, machineNums + 1))
>>> # Building a Qubo Model
>>> qubo_model = kw.core.QuboModel()
>>> X = kw.core.ndarray([taskNums, machineNums], "X", kw.core.Binary)
>>> J_mean = (np.sum(duration) + np.sum(machine_start_time)) / machineNums
>>> J = [machine_start_time[i] + duration.dot(X[:, i]) for i in range(machineNums)]
>>> # Set objective function
>>> qubo_model.set_objective(kw.core.quicksum([(J[i] - J_mean) ** 2 for i in range(machineNums)]) / machineNums)
>>> # Set constraint
>>> for j in range(taskNums):
...     qubo_model.add_constraint(kw.core.quicksum([X[j][i] for i in range(machineNums)]) == 1,
...                               f"c{j}", penalty=45)
>>> # Loop control
>>> controller = kw.common.SolverLoopController(max_repeat_step=5)
>>> # optimizer
>>> optimizer = kw.classical.SimulatedAnnealingOptimizer(initial_temperature=1e8,
...                                                      alpha=0.9,
...                                                      cutoff_temperature=0.01,
...                                                      iterations_per_t=200,
...                                                      size_limit=100)
>>> solver = kw.hybrid.PenaltyMethodOptimizer(optimizer, controller)
>>> sol_dict, hmt = solver.solve_qubo(qubo_model)
>>> J_end = np.zeros(machineNums)
>>> for i in range(machineNums):
...     ifturn_temp = kw.core.get_val(kw.core.quicksum(X[:, i].tolist()), sol_dict)
...     ifturn = 1 if ifturn_temp > 0 else 0
...     J_temp = duration.dot(X[:, i]) + machine_start_time[i] * ifturn
...     J_end[i] = kw.core.get_val(J_temp, sol_dict)
>>> print("duration array: ", duration)  
duration array:  [ 1  2  3  4  5  6  7  8  9 10 11 12 13 14 15 16 17 18 19 20]
>>> print("Machine startup time: ", machine_start_time)  
Machine startup time:  [1 2 3 4 5]
>>> print("End time of all machines:", J_end)  
End time of all machines: [43. 43. 48. 45. 46.]
>>> print('final result:{}'.format(np.max(J_end)))  
final result:48.0
>>> print('variance:', np.var(J_end))   
variance: 3.6
>>> kw.common.CheckpointManager.save_dir = None
solve_qubo(*args, **kwargs)#
solve_qubo_multi_results(qubo_model, size_limit=10)[source]#

Solve a QUBO model and return multiple solutions

Args:

qubo_model (QuboModel): QUBO model

size_limit (int): number of solutions to return

Returns:

list: dictionary of QUBO model solution SolutionResult and objective value

to_json_dict(exclude_fields=())[source]#

Convert to a JSON dictionary

Returns:

dict: JSON dictionary

load_json_dict(json_dict)[source]#

Restore an object from a dict read from a JSON file

Returns:

dict: JSON dictionary