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,JsonSerializableMixinPenalty 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)#