{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "%matplotlib inline" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "\n# Superheroes Factory\n\nThis example demostrates how to use the low-sugar in combination with mlflow\nto solve a simple supply chain problem.\n\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "# sphinx_gallery_thumbnail_path = '_static/superheroes.png'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Problem Description\n\nPopulate description here later.\n\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Let's start by doing some useful imports.\n\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "from itertools import product\nfrom typing import Dict\nimport gurobipy as gp" ] }, { "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": [ "def build_indices(data: Dict) -> Dict:\n customers = list(data[\"demand\"].keys())\n accessories = list(data[\"initial_inventory\"].keys())\n products = list(data[\"recipe\"].keys())\n max_day = max(demand_details[\"date\"] for demand_details in data[\"demand\"].values())\n days = range(1, max_day + 1)\n customer_dates = {\n (customer, day)\n for customer in customers\n for day in range(\n data[\"demand\"][customer][\"date\"],\n min(\n data[\"demand\"][customer][\"date\"] + 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\ndef build_variables(model: gp.Model, indices: Dict):\n dispatch = model.addVars(indices[\"customer_dates\"], vtype=\"B\", name=\"dispatch\")\n inventory = model.addVars(\n product(indices[\"accessories\"], indices[\"days\"]),\n vtype=\"C\",\n name=\"inventory\",\n )\n from_inventory = 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 = 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 = model.addVars(\n product(indices[\"accessories\"], indices[\"days\"]),\n vtype=\"C\",\n name=\"to_inventory\",\n )\n\n extra_production = 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 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 variables\n\n\ndef build_constraints(model: gp.Model, indices: Dict, variables: Dict) -> gp.Model:\n to_inventory = variables[\"to_inventory\"]\n from_factory = variables[\"from_factory\"]\n from_inventory = variables[\"from_inventory\"]\n inventory = variables[\"inventory\"]\n dispatch = variables[\"dispatch\"]\n extra_production = variables[\"extra_production\"]\n\n production = data[\"production\"]\n initial_inventory = data[\"initial_inventory\"]\n inventory_capacity = data[\"inventory_capacity\"]\n demand = data[\"demand\"]\n recipe = data[\"recipe\"]\n\n for accessory, day in product(indices[\"accessories\"], indices[\"days\"]):\n if str(day) in production[accessory]:\n 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 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(indices[\"accessories\"], indices[\"days\"]):\n if day == 1:\n model.addConstr(\n inventory[accessory, day]\n == initial_inventory[accessory]\n + to_inventory[accessory, day]\n - from_inventory.sum(\"*\", \"*\", accessory, day),\n name=f\"keeping_track_of_inventories_{accessory}_{day}\",\n )\n else:\n model.addConstr(\n inventory[accessory, day]\n == inventory[accessory, day - 1]\n + to_inventory[accessory, day]\n - from_inventory.sum(\"*\", \"*\", accessory, day),\n name=f\"keeping_track_of_inventories_{accessory}_{day}\",\n )\n\n for accessory, day in product(indices[\"accessories\"], indices[\"days\"]):\n model.addConstr(\n inventory[accessory, day] <= inventory_capacity[accessory],\n name=f\"inventory_capacity_{accessory}_{day}\",\n )\n\n for customer in indices[\"customers\"]:\n model.addConstr(\n dispatch.sum(customer, \"*\") == 1, name=f\"customer_served_{customer}\"\n )\n\n for prod, accessory in product(indices[\"products\"], indices[\"accessories\"]):\n for customer, day in indices[\"customer_dates\"]:\n 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 return model\n\n\ndef build_objective(\n model: gp.Model, indices: Dict, variables: Dict, demand: Dict\n) -> gp.Model:\n dispatch = variables[\"dispatch\"]\n extra_production = variables[\"extra_production\"]\n\n delay_penalty_costs = gp.quicksum(\n gp.quicksum(\n day * dispatch[customer, day]\n for customer, day in dispatch\n if customer == customer_\n )\n - demand[customer_][\"date\"]\n for customer_ in indices[\"customers\"]\n )\n\n cost_per_extra_prod = 2\n extra_production_costs = cost_per_extra_prod * extra_production.sum()\n\n total_cost = delay_penalty_costs + extra_production_costs\n\n model.setObjective(total_cost, gp.GRB.MINIMIZE)\n\n return model\n\n\ndef build(data: Dict) -> gp.Model:\n model = gp.Model(name=\"supply_chain_blended\")\n indices = build_indices(data)\n variables = build_variables(model, indices)\n model = build_constraints(model, indices, variables)\n model = build_objective(model, indices, variables, demand=data[\"demand\"])\n model.update()\n return model" ] }, { "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": [ "# import sys; sys.path.append(\"/Users/Juan.ChaconLeon/opt/opt-sugar/src\") # when running locally\nfrom opt_sugar import low_sugar\n\n# Setting the experiment\nimport datetime\nimport logging\n\nlogging.getLogger(\"mlflow\").setLevel(logging.CRITICAL) # Can be set DEBUG\n\nimport mlflow\nfrom mlflow import MlflowException\n\nexperiment_name = f\"superheros_blended_{datetime.datetime.now().strftime('%Y_%m_%d')}\"\ntry:\n experiment_id = mlflow.create_experiment(name=experiment_name)\nexcept MlflowException:\n experiment_id = mlflow.get_experiment_by_name(name=experiment_name).experiment_id" ] }, { "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": [ "import requests\nimport matplotlib.pyplot as plt\n\nwith mlflow.start_run(experiment_id=experiment_id):\n examples_path = (\n \"https://raw.githubusercontent.com/juandados/opt-sugar/main/examples/data\"\n )\n data = requests.get(f\"{examples_path}/supply_chain_blended_mini_toy.json\").json()\n\n opt_model = low_sugar.Model(build)\n result = opt_model.optimize(data=data)\n from utils.supply_chain import plot_supply_chain\n plot_supply_chain(data=data, solution=result['vars'])\n plt.show()\n model_info = mlflow.sklearn.log_model(opt_model, \"supply_chain_blended\")" ] } ], "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 }