lot_sizing_pipeline.py

              flowchart TD
              lot_sizing.lot_sizing_pipeline.GetCosts[GetCosts]

              

              click lot_sizing.lot_sizing_pipeline.GetCosts href "" "lot_sizing.lot_sizing_pipeline.GetCosts"
            

19
20
21
22
23
24
25
26
27
28
29
30

class GetCosts(CoSyLuigiTask):
    def output(self):
        return {"costs": luigi.LocalTarget("data/costs.json")}

    def run(self):
        d = {
            "fixedCost": 400,  # Bestellkosten
            "varCost": 1,  # Lagerhaltungssatz
        }
        os.makedirs("data", exist_ok=True)
        with open(self.output()["costs"].path, "w") as f:
            json.dump(d, f, indent=4)

20
21

def output(self):
    return {"costs": luigi.LocalTarget("data/costs.json")}

23
24
25
26
27
28
29
30

def run(self):
    d = {
        "fixedCost": 400,  # Bestellkosten
        "varCost": 1,  # Lagerhaltungssatz
    }
    os.makedirs("data", exist_ok=True)
    with open(self.output()["costs"].path, "w") as f:
        json.dump(d, f, indent=4)

              flowchart TD
              lot_sizing.lot_sizing_pipeline.GetHistoricDemand[GetHistoricDemand]

              

              click lot_sizing.lot_sizing_pipeline.GetHistoricDemand href "" "lot_sizing.lot_sizing_pipeline.GetHistoricDemand"
            

33
34
35
36
37
38
39

class GetHistoricDemand(CoSyLuigiTask):
    def output(self):
        return {"historic_demand": luigi.LocalTarget("data/historic_demand.csv")}

    def run(self):
        with self.output()["historic_demand"].open("w") as f:
            f.write("1, 5, 7, 8, 9, 10, 14, 16, 19, 21, 19, 23, 24, 26, 26, 26, 28, 26, 28, 30")

34
35

def output(self):
    return {"historic_demand": luigi.LocalTarget("data/historic_demand.csv")}

37
38
39

def run(self):
    with self.output()["historic_demand"].open("w") as f:
        f.write("1, 5, 7, 8, 9, 10, 14, 16, 19, 21, 19, 23, 24, 26, 26, 26, 28, 26, 28, 30")

              flowchart TD
              lot_sizing.lot_sizing_pipeline.PredictDemand[PredictDemand]

              

              click lot_sizing.lot_sizing_pipeline.PredictDemand href "" "lot_sizing.lot_sizing_pipeline.PredictDemand"
            

42
43
44
45
46
47
48

class PredictDemand(CoSyLuigiTask, ABC):
    get_historic_demand = CoSyLuigiTaskParameter(GetHistoricDemand)
    prediction_horizon = 8
    output_filename: str = ""

    def output(self):
        return {"predicted_demand": luigi.LocalTarget(self.output_filename)}

47
48

def output(self):
    return {"predicted_demand": luigi.LocalTarget(self.output_filename)}

              flowchart TD
              lot_sizing.lot_sizing_pipeline.PredictDemandByAverage[PredictDemandByAverage]
              lot_sizing.lot_sizing_pipeline.PredictDemand[PredictDemand]

                              lot_sizing.lot_sizing_pipeline.PredictDemand --> lot_sizing.lot_sizing_pipeline.PredictDemandByAverage
                


              click lot_sizing.lot_sizing_pipeline.PredictDemandByAverage href "" "lot_sizing.lot_sizing_pipeline.PredictDemandByAverage"
              click lot_sizing.lot_sizing_pipeline.PredictDemand href "" "lot_sizing.lot_sizing_pipeline.PredictDemand"
            

51
52
53
54
55
56
57
58
59
60
61
62
63

class PredictDemandByAverage(PredictDemand):
    output_filename = "data/predicted_demand_by_average.json"

    def run(self):
        with self.input()["get_historic_demand"]["historic_demand"].open() as infile:
            text = infile.read()
            historic_demand = [int(t) for t in text.split(",")]
            avg = int(sum(historic_demand) / len(historic_demand) + 0.5)
            predicted = [avg for _ in range(self.prediction_horizon)]
            data = {"predicted_demand": predicted}
            df_predicted = pd.DataFrame(data)

            df_predicted.to_json(self.output()["predicted_demand"].path)

54
55
56
57
58
59
60
61
62
63

def run(self):
    with self.input()["get_historic_demand"]["historic_demand"].open() as infile:
        text = infile.read()
        historic_demand = [int(t) for t in text.split(",")]
        avg = int(sum(historic_demand) / len(historic_demand) + 0.5)
        predicted = [avg for _ in range(self.prediction_horizon)]
        data = {"predicted_demand": predicted}
        df_predicted = pd.DataFrame(data)

        df_predicted.to_json(self.output()["predicted_demand"].path)

              flowchart TD
              lot_sizing.lot_sizing_pipeline.PredictDemandByLinearRegression[PredictDemandByLinearRegression]
              lot_sizing.lot_sizing_pipeline.PredictDemand[PredictDemand]

                              lot_sizing.lot_sizing_pipeline.PredictDemand --> lot_sizing.lot_sizing_pipeline.PredictDemandByLinearRegression
                


              click lot_sizing.lot_sizing_pipeline.PredictDemandByLinearRegression href "" "lot_sizing.lot_sizing_pipeline.PredictDemandByLinearRegression"
              click lot_sizing.lot_sizing_pipeline.PredictDemand href "" "lot_sizing.lot_sizing_pipeline.PredictDemand"
            

66
67
68
69
70
71
72
73
74
75
76

class PredictDemandByLinearRegression(PredictDemand):
    output_filename = "data/predicted_demand_by_linear_regression.json"

    def run(self):
        with self.input()["get_historic_demand"]["historic_demand"].open():
            # Mocked for sake of example
            predicted = [10 + i for i in range(self.prediction_horizon)]
            data = {"predicted_demand": predicted}
            df_predicted = pd.DataFrame(data)

            df_predicted.to_json(self.output()["predicted_demand"].path)

69
70
71
72
73
74
75
76

def run(self):
    with self.input()["get_historic_demand"]["historic_demand"].open():
        # Mocked for sake of example
        predicted = [10 + i for i in range(self.prediction_horizon)]
        data = {"predicted_demand": predicted}
        df_predicted = pd.DataFrame(data)

        df_predicted.to_json(self.output()["predicted_demand"].path)

              flowchart TD
              lot_sizing.lot_sizing_pipeline.OptimizeLots[OptimizeLots]

              

              click lot_sizing.lot_sizing_pipeline.OptimizeLots href "" "lot_sizing.lot_sizing_pipeline.OptimizeLots"
            

 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112

class OptimizeLots(CoSyLuigiTask, ABC):
    predict_demand = CoSyLuigiTaskParameter(PredictDemand)
    get_costs = CoSyLuigiTaskParameter(GetCosts)
    output_filename: str = ""

    def _get_cost(self):
        with open(self.input()["get_costs"]["costs"].path, "rb") as f:
            return json.load(f)

    def _get_demand(self):
        demand_df = pd.read_json(self.input()["predict_demand"]["predicted_demand"].path)
        return list(demand_df["predicted_demand"])

    def output(self):
        return {"optimized_lots": luigi.LocalTarget("data/" + self._get_variant_label() + "-" + self.output_filename)}

    def run(self):
        cost = self._get_cost()
        demand = self._get_demand()

        orders = self.run_optimizer(cost, demand)

        with self.output()["optimized_lots"].open("w") as f:
            f.write(str(list(orders)))

    @abstractmethod
    def run_optimizer(self, cost, demand):
        return NotImplementedError()

    def _get_variant_label(self):
        if isinstance(self.input()["predict_demand"]["predicted_demand"], luigi.LocalTarget):
            label = self.input()["predict_demand"]["predicted_demand"].path
            return Path(label).stem
        return None

92
93

def output(self):
    return {"optimized_lots": luigi.LocalTarget("data/" + self._get_variant_label() + "-" + self.output_filename)}

 95
 96
 97
 98
 99
100
101
102

def run(self):
    cost = self._get_cost()
    demand = self._get_demand()

    orders = self.run_optimizer(cost, demand)

    with self.output()["optimized_lots"].open("w") as f:
        f.write(str(list(orders)))

104
105
106

@abstractmethod
def run_optimizer(self, cost, demand):
    return NotImplementedError()

              flowchart TD
              lot_sizing.lot_sizing_pipeline.OptimizeLotsByGroff[OptimizeLotsByGroff]
              lot_sizing.lot_sizing_pipeline.OptimizeLots[OptimizeLots]

                              lot_sizing.lot_sizing_pipeline.OptimizeLots --> lot_sizing.lot_sizing_pipeline.OptimizeLotsByGroff
                


              click lot_sizing.lot_sizing_pipeline.OptimizeLotsByGroff href "" "lot_sizing.lot_sizing_pipeline.OptimizeLotsByGroff"
              click lot_sizing.lot_sizing_pipeline.OptimizeLots href "" "lot_sizing.lot_sizing_pipeline.OptimizeLots"
            

115
116
117
118
119
120

class OptimizeLotsByGroff(OptimizeLots):
    output_filename = "optimize_lots_by_groff.txt"

    def run_optimizer(self, cost, demand):
        optimizer = GroffHeuristic()
        return optimizer.run(cost, demand)

118
119
120

def run_optimizer(self, cost, demand):
    optimizer = GroffHeuristic()
    return optimizer.run(cost, demand)

              flowchart TD
              lot_sizing.lot_sizing_pipeline.OptimizeLotsByWagnerWhitin[OptimizeLotsByWagnerWhitin]
              lot_sizing.lot_sizing_pipeline.OptimizeLots[OptimizeLots]

                              lot_sizing.lot_sizing_pipeline.OptimizeLots --> lot_sizing.lot_sizing_pipeline.OptimizeLotsByWagnerWhitin
                


              click lot_sizing.lot_sizing_pipeline.OptimizeLotsByWagnerWhitin href "" "lot_sizing.lot_sizing_pipeline.OptimizeLotsByWagnerWhitin"
              click lot_sizing.lot_sizing_pipeline.OptimizeLots href "" "lot_sizing.lot_sizing_pipeline.OptimizeLots"
            

123
124
125
126
127
128

class OptimizeLotsByWagnerWhitin(OptimizeLots):
    output_filename = "optimize_lots_by_wagner_within.txt"

    def run_optimizer(self, cost, demand):
        optimizer = WagnerWhitin()
        return optimizer.run(cost, demand)

126
127
128

def run_optimizer(self, cost, demand):
    optimizer = WagnerWhitin()
    return optimizer.run(cost, demand)

              flowchart TD
              lot_sizing.lot_sizing_pipeline.OptimizeLotsBySilverMeal[OptimizeLotsBySilverMeal]
              lot_sizing.lot_sizing_pipeline.OptimizeLots[OptimizeLots]

                              lot_sizing.lot_sizing_pipeline.OptimizeLots --> lot_sizing.lot_sizing_pipeline.OptimizeLotsBySilverMeal
                


              click lot_sizing.lot_sizing_pipeline.OptimizeLotsBySilverMeal href "" "lot_sizing.lot_sizing_pipeline.OptimizeLotsBySilverMeal"
              click lot_sizing.lot_sizing_pipeline.OptimizeLots href "" "lot_sizing.lot_sizing_pipeline.OptimizeLots"
            

131
132
133
134
135
136

class OptimizeLotsBySilverMeal(OptimizeLots):
    output_filename = "optimize_lots_by_silver_meal.txt"

    def run_optimizer(self, cost, demand):
        optimizer = SilverMeal()
        return optimizer.run(cost, demand)

134
135
136

def run_optimizer(self, cost, demand):
    optimizer = SilverMeal()
    return optimizer.run(cost, demand)

              flowchart TD
              lot_sizing.lot_sizing_pipeline.OptimizeLotsByLeastUnitCost[OptimizeLotsByLeastUnitCost]
              lot_sizing.lot_sizing_pipeline.OptimizeLots[OptimizeLots]

                              lot_sizing.lot_sizing_pipeline.OptimizeLots --> lot_sizing.lot_sizing_pipeline.OptimizeLotsByLeastUnitCost
                


              click lot_sizing.lot_sizing_pipeline.OptimizeLotsByLeastUnitCost href "" "lot_sizing.lot_sizing_pipeline.OptimizeLotsByLeastUnitCost"
              click lot_sizing.lot_sizing_pipeline.OptimizeLots href "" "lot_sizing.lot_sizing_pipeline.OptimizeLots"
            

139
140
141
142
143
144

class OptimizeLotsByLeastUnitCost(OptimizeLots):
    output_filename = "optimize_lots_by_least_unit_cost.txt"

    def run_optimizer(self, cost, demand):
        optimizer = LeastUnitCostMethod()
        return optimizer.run(cost, demand)

142
143
144

def run_optimizer(self, cost, demand):
    optimizer = LeastUnitCostMethod()
    return optimizer.run(cost, demand)

              flowchart TD
              lot_sizing.lot_sizing_pipeline.OptimizeLotsByPartPeriod[OptimizeLotsByPartPeriod]
              lot_sizing.lot_sizing_pipeline.OptimizeLots[OptimizeLots]

                              lot_sizing.lot_sizing_pipeline.OptimizeLots --> lot_sizing.lot_sizing_pipeline.OptimizeLotsByPartPeriod
                


              click lot_sizing.lot_sizing_pipeline.OptimizeLotsByPartPeriod href "" "lot_sizing.lot_sizing_pipeline.OptimizeLotsByPartPeriod"
              click lot_sizing.lot_sizing_pipeline.OptimizeLots href "" "lot_sizing.lot_sizing_pipeline.OptimizeLots"
            

147
148
149
150
151
152

class OptimizeLotsByPartPeriod(OptimizeLots):
    output_filename = "optimize_lots_by_part_period.txt"

    def run_optimizer(self, cost, demand):
        optimizer = PartPeriod()
        return optimizer.run(cost, demand)

150
151
152

def run_optimizer(self, cost, demand):
    optimizer = PartPeriod()
    return optimizer.run(cost, demand)