notebooks/symk_usage.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e3cf0883",
   "metadata": {},
   "source": [
    "# Multi-Solution Generation: Using SymK in the Unified Planning Library"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7b6ae544",
   "metadata": {},
   "source": [
    "## A simple example\n",
    "\n",
    "We use the iconic problem number one of the gripper domain, where the goal is to move four balls from room A to room B with a gripper that has two arms."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ebb12ab8",
   "metadata": {},
   "source": [
    "To utilize the unified-planning library and SymK, it is necessary to install them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2dcec7e3",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!pip install --pre unified-planning\n",
    "!pip install up-symk"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "54204a66",
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "from collections import defaultdict\n",
    "\n",
    "import unified_planning as up\n",
    "import up_symk\n",
    "from unified_planning.shortcuts import *\n",
    "from unified_planning.io import PDDLReader"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fbf3f019",
   "metadata": {},
   "source": [
    "## Domain and Problem Files\n",
    "We need to download the domain and problem files described in PDDL for the planning task we aim to solve. Following this, we read and use these files to create a PDDL problem, setting the quality metric to minimize the plan length."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "62b37615",
   "metadata": {},
   "outputs": [],
   "source": [
    "!wget https://raw.githubusercontent.com/speckdavid/up-symk/master/notebooks/gripper-domain.pddl\n",
    "!wget https://raw.githubusercontent.com/speckdavid/up-symk/master/notebooks/gripper-prob01.pddl"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "db40758b",
   "metadata": {},
   "outputs": [],
   "source": [
    "reader = PDDLReader()\n",
    "pddl_problem = reader.parse_problem(\"gripper-domain.pddl\", \"gripper-prob01.pddl\")\n",
    "pddl_problem.add_quality_metric(MinimizeSequentialPlanLength())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ab964b3",
   "metadata": {},
   "source": [
    "## Validator\n",
    "We define two functions that serve as helper functions to evaluate the found plans and calculate the cost of the plans."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "9f025ef9",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_plan_cost(problem, plan):\n",
    "    pv = PlanValidator(problem_kind=problem.kind)\n",
    "    pv_res = pv.validate(problem, plan)\n",
    "    return pv_res.metric_evaluations[problem.quality_metrics[0]]\n",
    "\n",
    "def get_plans_by_cost(problem, plans):\n",
    "    pv = PlanValidator(problem_kind=problem.kind)\n",
    "    plans_by_cost = defaultdict(lambda: [])\n",
    "    for plan in plans:\n",
    "        pv_res = pv.validate(problem, plan)\n",
    "        cost = pv_res.metric_evaluations[problem.quality_metrics[0]]\n",
    "        plans_by_cost[cost].append(result.plan)\n",
    "    return plans_by_cost"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fb03e2ef",
   "metadata": {},
   "source": [
    "## Finding a single optimal solution\n",
    "We solve our gripper problem optimally with ``symk-opt``."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "23a8eefa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[96m\u001b[1mNOTE: To disable printing of planning engine credits, add this line to your code: `up.shortcuts.get_environment().credits_stream = None`\n",
      "\u001b[0m\u001b[96m  *** Credits ***\n",
      "\u001b[0m\u001b[96m  * In operation mode `OneshotPlanner` at line 1 of `/tmp/ipykernel_39155/3917870263.py`, \u001b[0m\u001b[96myou are using the following planning engine:\n",
      "\u001b[0m\u001b[96m  * Engine name: SymK\n",
      "  * Developers:  David Speck (cf. https://github.com/speckdavid/symk/blob/master/README.md )\n",
      "\u001b[0m\u001b[96m  * Description: \u001b[0m\u001b[96mSymK is a state-of-the-art domain-independent optimal and top-k planner.\u001b[0m\u001b[96m\n",
      "\u001b[0m\u001b[96m\n",
      "\u001b[0mSymK (with optimality guarantee) found this plan: SequentialPlan:\n",
      "    pick(ball2, rooma, right)\n",
      "    pick(ball1, rooma, left)\n",
      "    move(rooma, roomb)\n",
      "    drop(ball2, roomb, right)\n",
      "    drop(ball1, roomb, left)\n",
      "    move(roomb, rooma)\n",
      "    pick(ball3, rooma, left)\n",
      "    pick(ball4, rooma, right)\n",
      "    move(rooma, roomb)\n",
      "    drop(ball3, roomb, left)\n",
      "    drop(ball4, roomb, right) with cost 11.\n"
     ]
    }
   ],
   "source": [
    "with OneshotPlanner(name='symk-opt') as planner:\n",
    "    result = planner.solve(pddl_problem) # output_stream=sys.stdout\n",
    "    if result.status in unified_planning.engines.results.POSITIVE_OUTCOMES:\n",
    "        cost = get_plan_cost(pddl_problem, result.plan)\n",
    "        print(f\"{planner.name} found this plan: {result.plan} with cost {cost}.\")\n",
    "    else:\n",
    "        print(\"No plan found.\")\n",
    "        \n",
    "# We do not want to see the credits again and again, so let's disable them from now on.\n",
    "up.shortcuts.get_environment().credits_stream = None"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "466266be",
   "metadata": {},
   "source": [
    "## Finding multiple optimal solutions\n",
    "We solve our gripper problem with ``symk-opt`` and ask for three optimal solutions, which are reported."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "8d28bee7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Plan 1: SequentialPlan:\n",
      "    pick(ball3, rooma, right)\n",
      "    pick(ball2, rooma, left)\n",
      "    move(rooma, roomb)\n",
      "    drop(ball3, roomb, right)\n",
      "    drop(ball2, roomb, left)\n",
      "    move(roomb, rooma)\n",
      "    pick(ball1, rooma, left)\n",
      "    pick(ball4, rooma, right)\n",
      "    move(rooma, roomb)\n",
      "    drop(ball1, roomb, left)\n",
      "    drop(ball4, roomb, right) with cost 11.\n",
      "\n",
      "Plan 2: SequentialPlan:\n",
      "    pick(ball3, rooma, right)\n",
      "    pick(ball2, rooma, left)\n",
      "    move(rooma, roomb)\n",
      "    drop(ball3, roomb, right)\n",
      "    drop(ball2, roomb, left)\n",
      "    move(roomb, rooma)\n",
      "    pick(ball1, rooma, left)\n",
      "    pick(ball4, rooma, right)\n",
      "    move(rooma, roomb)\n",
      "    drop(ball4, roomb, right)\n",
      "    drop(ball1, roomb, left) with cost 11.\n",
      "\n",
      "Plan 3: SequentialPlan:\n",
      "    pick(ball2, rooma, left)\n",
      "    pick(ball3, rooma, right)\n",
      "    move(rooma, roomb)\n",
      "    drop(ball3, roomb, right)\n",
      "    drop(ball2, roomb, left)\n",
      "    move(roomb, rooma)\n",
      "    pick(ball1, rooma, left)\n",
      "    pick(ball4, rooma, right)\n",
      "    move(rooma, roomb)\n",
      "    drop(ball1, roomb, left)\n",
      "    drop(ball4, roomb, right) with cost 11.\n",
      "\n",
      "SymK (with optimality guarantee) found 3 optimal plans with cost 11.\n"
     ]
    }
   ],
   "source": [
    "plans = []\n",
    "\n",
    "with AnytimePlanner(name='symk-opt', params={\"number_of_plans\": 3}) as planner:\n",
    "    for i, result in enumerate(planner.get_solutions(pddl_problem)): # output_stream=sys.stdout): \n",
    "        if result.status == up.engines.PlanGenerationResultStatus.INTERMEDIATE:\n",
    "            plans.append(result.plan)\n",
    "            print(f\"Plan {i+1}: {result.plan} with cost {get_plan_cost(pddl_problem, result.plan)}.\")\n",
    "            print()\n",
    "        elif result.status in unified_planning.engines.results.POSITIVE_OUTCOMES:\n",
    "            for cost, plans in get_plans_by_cost(pddl_problem, plans).items():\n",
    "                print(f\"{planner.name} found {len(plans)} optimal plans with cost {cost}.\")\n",
    "        elif result.status not in unified_planning.engines.results.POSITIVE_OUTCOMES:\n",
    "            print(\"No plan found.\") "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a49bd090",
   "metadata": {},
   "source": [
    "## Finding all optimal solutions\n",
    "We find all optimal solutions (using ``symk-opt``) and report the number of existing optimal plans and their cost. Note that by default the number of schedules generated in Anytime mode is 1, so you need to specify the number (None sets it to infinite)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "69c01782",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SymK (with optimality guarantee)\n",
      "SymK (with optimality guarantee) found 100 plans...\n",
      "SymK (with optimality guarantee) found 200 plans...\n",
      "SymK (with optimality guarantee) found 300 plans...\n",
      "\n",
      "SymK (with optimality guarantee) found 384 plans!\n",
      "\n",
      "Calculate the cost of the plans found...\n",
      "SymK (with optimality guarantee) found 384 plans with cost 11.\n"
     ]
    }
   ],
   "source": [
    "plans = []\n",
    "\n",
    "with AnytimePlanner(problem_kind=pddl_problem.kind, anytime_guarantee=\"OPTIMAL_PLANS\", \n",
    "                    params={\"number_of_plans\": None}) as planner:\n",
    "    print(planner.name)\n",
    "    for i, result in enumerate(planner.get_solutions(pddl_problem)): # output_stream=sys.stdout): \n",
    "        if result.status == up.engines.PlanGenerationResultStatus.INTERMEDIATE:\n",
    "            plans.append(result.plan)\n",
    "            if i > 0 and i % 100 == 0:\n",
    "                print(f\"{planner.name} found {i} plans...\")\n",
    "        elif result.status in unified_planning.engines.results.POSITIVE_OUTCOMES:\n",
    "            print()\n",
    "            print(f\"{planner.name} found {i} plans!\")\n",
    "            print()\n",
    "        elif result.status not in unified_planning.engines.results.POSITIVE_OUTCOMES:\n",
    "            print(\"No plan found.\")\n",
    "            \n",
    "# Calculate the cost of the plans found\n",
    "print(\"Calculate the cost of the plans found...\")\n",
    "for cost, plans in get_plans_by_cost(pddl_problem, plans).items():\n",
    "    print(f\"{planner.name} found {len(plans)} plans with cost {cost}.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6c175250",
   "metadata": {},
   "source": [
    "## Finding multiple solutions ordered by cost\n",
    "Again, we solve our grab problem and this time we ask for 500 plans, of which 384 plans are optimal with cost 11 and the remaining 116 of the 500 plans have cost 12."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "253b5799",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SymK found 100 plans...\n",
      "SymK found 200 plans...\n",
      "SymK found 300 plans...\n",
      "SymK found 400 plans...\n",
      "\n",
      "SymK found 500 plans!\n",
      "\n",
      "Calculate the cost of the plans found...\n",
      "SymK found 384 plans with cost 11.\n",
      "SymK found 116 plans with cost 12.\n"
     ]
    }
   ],
   "source": [
    "# Find 500 plans with SymK\n",
    "plans = []\n",
    "\n",
    "with AnytimePlanner(name='symk', params={\"number_of_plans\": 500}) as planner:\n",
    "    for i, result in enumerate(planner.get_solutions(pddl_problem)): # output_stream=sys.stdout): \n",
    "        if result.status == up.engines.PlanGenerationResultStatus.INTERMEDIATE:\n",
    "            plans.append(result.plan)\n",
    "            if i > 0 and i % 100 == 0:\n",
    "                print(f\"{planner.name} found {i} plans...\")\n",
    "        elif result.status in unified_planning.engines.results.POSITIVE_OUTCOMES:\n",
    "            print()\n",
    "            print(f\"{planner.name} found {i} plans!\")\n",
    "            print()\n",
    "        elif result.status not in unified_planning.engines.results.POSITIVE_OUTCOMES:\n",
    "            print(\"No plan found.\")\n",
    "\n",
    "# Calculate the cost of the plans found\n",
    "print(\"Calculate the cost of the plans found...\")\n",
    "for cost, plans in get_plans_by_cost(pddl_problem, plans).items():\n",
    "    print(f\"{planner.name} found {len(plans)} plans with cost {cost}.\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "eubug",
   "language": "python",
   "name": "eubug"
  },
  "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.18"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}