Coverage for moptipyapps / prodsched / rop_experiment.py: 33%

1""" 

2A small template for ROP-based experiments. 

3 

4This experiment uses the NSGA-II algorithm to optimize Re-Order-Points (ROPs) 

5to achieve both a high worst-case fillrate and a low worst-case average stock 

6level. 

7""" 

8 

9 

10import argparse 

11from typing import Final 

12 

13from moptipy.algorithms.mo.nsga2 import NSGA2 

14from moptipy.api.experiment import run_experiment 

15from moptipy.api.mo_execution import MOExecution 

16from moptipy.mo.problem.weighted_sum import WeightedSum 

17from moptipy.operators.intspace.op0_random import Op0Random 

18from moptipy.operators.intspace.op1_mnormal import Op1MNormal 

19from moptipy.operators.intspace.op2_uniform import Op2Uniform 

20from moptipy.spaces.intspace import IntSpace 

21from pycommons.io.console import logger 

22from pycommons.io.path import Path 

23from pycommons.types import check_int_range 

24 

25from moptipyapps.prodsched.instance import Instance 

26from moptipyapps.prodsched.instances import get_instances 

27from moptipyapps.prodsched.multistatistics import MultiStatisticsSpace 

28from moptipyapps.prodsched.objectives.max_stocklevel import MaxStockLevel 

29from moptipyapps.prodsched.objectives.worst_and_mean_fill_rate import ( 

30 WorstAndMeanFillRate, 

31) 

32from moptipyapps.prodsched.rop_multisimulation import ROPMultiSimulation 

33from moptipyapps.utils.shared import moptipyapps_argparser 

34 

35 

36def run(dest: str, instances: str, n_inst: int, n_runs: int, 

37 max_fes: int, ps: int) -> None: 

38 """ 

39 Run the experiment. 

40 

41 :param dest: the destination directory 

42 :param instances: the directory with the instances 

43 :param n_inst: the number of instances 

44 :param n_runs: the number of runs 

45 :param max_fes: the maximum FEs 

46 :param ps: the population size 

47 """ 

48 logger(f"Beginning experiment with dest={dest!r}, instances={instances!r}" 

49 f", n_inst={n_inst}, n_runs={n_runs}, and max_fes={max_fes}.") 

50 use_dest: Final[Path] = Path(dest) 

51 use_dest.ensure_dir_exists() 

52 logger(f"Destination folder is {use_dest!r}.") 

53 

54 use_insts: Final[Path] = Path(instances) 

55 use_insts.ensure_dir_exists() 

56 logger(f"Instances folder is {use_insts!r}.") 

57 

58 check_int_range(n_inst, "n_inst", 1, 128) 

59 check_int_range(max_fes, "max_fes", 10, 10 ** 10) 

60 check_int_range(ps, "ps", 4, 16384) 

61 

62 logger(f"Loading {n_inst} instances from {use_insts!r}.") 

63 insts: Final[tuple[Instance, ...]] = get_instances(n_inst, instances) 

64 if tuple.__len__(insts) != n_inst: 

65 raise ValueError("Could not load required instances.") 

66 logger(f"Loaded {n_inst} instances from {use_insts!r}.") 

67 

68 n_prod: int | None = None 

69 for inst in insts: 

70 if n_prod is None: 

71 n_prod = inst.n_products 

72 elif n_prod != inst.n_products: 

73 raise ValueError("Inconsistent number of products!") 

74 if n_prod is None: 

75 raise ValueError("No instances?") 

76 

77 search_space: Final[IntSpace] = IntSpace(n_prod, 0, 63) 

78 op0: Final[Op0Random] = Op0Random(search_space) 

79 op1: Final[Op1MNormal] = Op1MNormal(search_space, sd=2.5) 

80 op2: Final = Op2Uniform() 

81 algo: Final[NSGA2] = NSGA2(op0, op1, op2, ps, 1 / min(16, ps)) 

82 encoding: Final[ROPMultiSimulation] = ROPMultiSimulation(insts) 

83 f1: Final[WorstAndMeanFillRate] = WorstAndMeanFillRate() 

84 f2: Final[MaxStockLevel] = MaxStockLevel() 

85 ws: Final[WeightedSum] = WeightedSum((f1, f2), ( 

86 (1 / (f1.upper_bound() - f1.lower_bound())), 1 / (2 * n_prod))) 

87 solution_space: Final[MultiStatisticsSpace] = MultiStatisticsSpace(insts) 

88 

89 def __setup(_) -> MOExecution: 

90 """ 

91 Set up the experiment. 

92 

93 :return: the execution 

94 """ 

95 return (MOExecution() 

96 .set_search_space(search_space) 

97 .set_algorithm(algo) 

98 .set_solution_space(solution_space) 

99 .set_objective(ws) 

100 .set_encoding(encoding) 

101 .set_max_fes(max_fes) 

102 .set_log_improvements(True)) 

103 

104 run_experiment(base_dir=use_dest, instances=(lambda: "all", ), 

105 setups=(__setup, ), n_runs=n_runs, 

106 pre_warmup_fes=2, perform_warmup=False, 

107 perform_pre_warmup=True) 

108 

109 

110# Run the experiment from the command line 

111if __name__ == "__main__": 

112 parser: Final[argparse.ArgumentParser] = moptipyapps_argparser( 

113 __file__, "ROP-based MFC Optimization", 

114 "Run a small experiment with ROP-based MFC optimization.") 

115 parser.add_argument( 

116 "dest", help="the directory to store the experimental results under", 

117 type=Path, nargs="?", default="./results/") 

118 parser.add_argument( 

119 "insts", help="the directory with the instances", 

120 type=Path, nargs="?", default="./instances/") 

121 parser.add_argument( 

122 "n_inst", help="the number of instances", 

123 type=int, nargs="?", default=11) 

124 parser.add_argument( 

125 "n_runs", help="the number of runs", 

126 type=int, nargs="?", default=31) 

127 parser.add_argument( 

128 "max_fes", help="the number of FEs per run", 

129 type=int, nargs="?", default=8192) 

130 parser.add_argument( 

131 "ps", help="the population size of NSGA-II", 

132 type=int, nargs="?", default=64) 

133 args: Final[argparse.Namespace] = parser.parse_args() 

134 run(args.dest, args.insts, args.n_inst, args.n_runs, args.max_fes, 

135 args.ps)