{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "%matplotlib inline" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "\n# Experiment Tracking: Coloring [IN PROGRESS!!!]\n\nThis example demostrates how to use opt-sugar in combination with mlflow\nfor single objective optimization experiment tracking\n\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "from itertools import product\nimport logging\nimport json\nimport pathlib\nimport datetime\n\nimport mlflow\nfrom mlflow.exceptions import MlflowException\n\nimport gurobipy as gp\n# import sys; sys.path.append('/Users/Juan.ChaconLeon/opt/opt-sugar/src') # when running locally\nfrom opt_sugar.extra_sugar import OptModel, ModelBuilder\nfrom opt_sugar.extra_sugar.objective import Objective, ObjectivePart, BaseObjective\n\n\nclass SupplyChainBlendedModelBuilder(ModelBuilder):\n \"\"\"This should be user implemented\"\"\"\n\n def __init__(self, data):\n super().__init__(data)\n self.variables = None\n self.indices = self._build_indices()\n\n def _build_indices(self):\n customers = list(self.data[\"demand\"].keys())\n accessories = list(self.data[\"initial_inventory\"].keys())\n products = list(self.data[\"recipe\"].keys())\n max_day = max(\n demand_details[\"date\"] for demand_details in self.data[\"demand\"].values()\n )\n days = range(1, max_day + 1)\n customer_dates = {\n (customer, day)\n for customer in customers\n for day in range(\n self.data[\"demand\"][customer][\"date\"],\n min(\n self.data[\"demand\"][customer][\"date\"] + self.data[\"max_delay\"],\n max_day + 1,\n ),\n )\n }\n\n indices = {\n \"customers\": customers,\n \"customer_dates\": customer_dates,\n \"max_day\": max_day,\n \"days\": days,\n \"accessories\": accessories,\n \"products\": products,\n }\n return indices\n\n def build_variables(self, base_model):\n indices = self.indices\n dispatch = base_model.addVars(\n indices[\"customer_dates\"], vtype=\"B\", name=\"dispatch\"\n )\n inventory = base_model.addVars(\n product(indices[\"accessories\"], indices[\"days\"]),\n vtype=\"C\",\n name=\"inventory\",\n )\n from_inventory = base_model.addVars(\n product(\n indices[\"customers\"],\n indices[\"products\"],\n indices[\"accessories\"],\n indices[\"days\"],\n ),\n vtype=\"I\",\n name=\"from_inventory\",\n )\n from_factory = base_model.addVars(\n product(\n indices[\"customers\"],\n indices[\"products\"],\n indices[\"accessories\"],\n indices[\"days\"],\n ),\n vtype=\"I\",\n name=\"from_factory\",\n )\n to_inventory = base_model.addVars(\n product(indices[\"accessories\"], indices[\"days\"]),\n vtype=\"C\",\n name=\"to_inventory\",\n )\n\n extra_production = base_model.addVars(\n product(\n indices[\"customers\"],\n indices[\"products\"],\n indices[\"accessories\"],\n indices[\"days\"],\n ),\n vtype=\"C\",\n name=\"extra_production\",\n )\n\n self.variables = {\n \"dispatch\": dispatch,\n \"inventory\": inventory,\n \"from_inventory\": from_inventory,\n \"from_factory\": from_factory,\n \"to_inventory\": to_inventory,\n \"extra_production\": extra_production,\n }\n return self.variables\n\n def build_constraints(self, base_model):\n accessories = self.indices[\"accessories\"]\n days = self.indices[\"days\"]\n customers = self.indices[\"customers\"]\n products = self.indices[\"products\"]\n customer_dates = self.indices[\"customer_dates\"]\n\n to_inventory = self.variables[\"to_inventory\"]\n from_factory = self.variables[\"from_factory\"]\n from_inventory = self.variables[\"from_inventory\"]\n inventory = self.variables[\"inventory\"]\n dispatch = self.variables[\"dispatch\"]\n extra_production = self.variables[\"extra_production\"]\n\n production = self.data[\"production\"]\n initial_inventory = self.data[\"initial_inventory\"]\n inventory_capacity = self.data[\"inventory_capacity\"]\n demand = self.data[\"demand\"]\n recipe = self.data[\"recipe\"]\n\n for accessory, day in product(accessories, days):\n if str(day) in production[accessory]:\n base_model.addConstr(\n to_inventory[accessory, day]\n + from_factory.sum(\"*\", \"*\", accessory, day)\n == production[accessory][str(day)],\n name=f\"production_allocation_{accessory}_{day}\",\n )\n else:\n base_model.addConstr(\n to_inventory[accessory, day]\n + from_factory.sum(\"*\", \"*\", accessory, day)\n == 0,\n name=f\"production_allocation_{accessory}_{day}\",\n )\n\n for accessory, day in product(accessories, days):\n if day == 1:\n base_model.addConstr(\n inventory[accessory, day]\n == initial_inventory[accessory] + to_inventory[accessory, day] - from_inventory.sum(\"*\", \"*\", accessory, day),\n name=f\"keeping_track_of_inventories_{accessory}_{day}\",\n )\n else:\n base_model.addConstr(\n inventory[accessory, day]\n == inventory[accessory, day - 1] + to_inventory[accessory, day] - from_inventory.sum(\"*\", \"*\", accessory, day),\n name=f\"keeping_track_of_inventories_{accessory}_{day}\",\n )\n\n for accessory, day in product(accessories, days):\n base_model.addConstr(\n inventory[accessory, day] <= inventory_capacity[accessory],\n name=f\"inventory_capacity_{accessory}_{day}\",\n )\n\n for customer in customers:\n base_model.addConstr(\n dispatch.sum(customer, \"*\") == 1, name=f\"customer_served_{customer}\"\n )\n\n for prod, accessory in product(products, accessories):\n for customer, day in customer_dates:\n base_model.addConstr(\n from_inventory[customer, prod, accessory, day]\n + from_factory[customer, prod, accessory, day]\n + extra_production[customer, prod, accessory, day]\n == dispatch[customer, day]\n * demand[customer][prod]\n * recipe[prod][accessory],\n name=f\"demand_satisfaction_{customer}_{prod}_{accessory}_{day}\",\n )\n\n def build_objective(self, base_model):\n dispatch = self.variables[\"dispatch\"]\n extra_production = self.variables[\"extra_production\"]\n customers = self.indices[\"customers\"]\n demand = self.data[\"demand\"]\n\n delay_penalty_costs = gp.quicksum(\n gp.quicksum(\n day * dispatch[customer, day]\n for customer, day in dispatch\n if customers == customer_\n )\n - demand[customer_][\"date\"]\n for customer_ in customers\n )\n extra_production_costs = 2 * extra_production.sum()\n\n objective_parts = [\n ObjectivePart(weight=0.5, expr=delay_penalty_costs),\n ObjectivePart(weight=0.5, expr=extra_production_costs),\n ]\n\n objective = Objective([BaseObjective(objective_parts, hierarchy=1)])\n base_model.setObjective(objective.build()[0], gp.GRB.MINIMIZE)\n return objective" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Tracking an Optimization Experiment\n\nAdd description here.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "logging.getLogger(\"mlflow\").setLevel(logging.CRITICAL) # Can be set DEBUG\n\ntry:\n experiment_name = f\"opt_exp_{datetime.datetime.now().strftime('%Y_%m_%d')}\"\n experiment_id = mlflow.create_experiment(name=experiment_name)\nexcept MlflowException:\n experiment_id = mlflow.get_experiment_by_name(name=experiment_name).experiment_id\n\n\nbase_path = pathlib.Path('.')\nexamples_path = base_path.parent / \"data\"\n\nwith open(examples_path / \"supply_chain_blended_mini_toy.json\", \"r\", encoding=\"utf-8\") as file:\n data = json.load(file)\n\nwith mlflow.start_run(experiment_id=experiment_id):\n opt_model = OptModel(model_builder=SupplyChainBlendedModelBuilder)\n solution = opt_model.optimize(data)" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.9.15" } }, "nbformat": 4, "nbformat_minor": 0 }