lab/02_TrainYourModel/02_Check Progress and Test the endpoint.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Now let's monitor the training/deploying process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import ipywidgets as widgets\n",
    "import time\n",
    "\n",
    "from IPython.display import display"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Helper functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_actions():\n",
    "    actions = []\n",
    "    executionId = None\n",
    "    resp = codepipeline.get_pipeline_state( name=pipeline_name )\n",
    "    for stage in resp['stageStates']:\n",
    "        stageName = stage['stageName']\n",
    "        stageStatus = None\n",
    "        if stage.get('latestExecution') is not None:\n",
    "            stageStatus = stage['latestExecution']['status']\n",
    "            if executionId is None:\n",
    "                executionId = stage['latestExecution']['pipelineExecutionId']\n",
    "            elif stage['latestExecution']['pipelineExecutionId'] != executionId:\n",
    "                stageStatus = 'Old'\n",
    "        for action in stage['actionStates']:\n",
    "            actionName = action['actionName']\n",
    "            actionStatus = 'Old'\n",
    "            if action.get('latestExecution') is not None and stageStatus != 'Old':\n",
    "                actionStatus = action['latestExecution']['status']\n",
    "            actions.append( {'stageName': stageName, \n",
    "                             'stageStatus': stageStatus, \n",
    "                             'actionName': actionName, \n",
    "                             'actionStatus': actionStatus})\n",
    "    return actions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_approval_token():\n",
    "    resp = codepipeline.get_pipeline_state( name=pipeline_name )\n",
    "    token = None\n",
    "    # Get the approve train status token\n",
    "    for stageState in resp['stageStates']:\n",
    "        if stageState['stageName'] == 'DeployApproval':\n",
    "            for actionState in stageState['actionStates']:\n",
    "                if actionState['actionName'] == 'ApproveDeploy':\n",
    "                    if actionState.get('latestExecution') is None:\n",
    "                        return None\n",
    "                    latestExecution = actionState['latestExecution']\n",
    "                    if latestExecution['status'] == 'InProgress':\n",
    "                        token = latestExecution['token']\n",
    "    return token"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.serializers import CSVSerializer\n",
    "csv_serializer = CSVSerializer()\n",
    "def test_endpoint(endpoint_name, payload):\n",
    "    resp = sm.invoke_endpoint(\n",
    "        EndpointName=endpoint_name,\n",
    "        ContentType='text/csv',\n",
    "        Accept='text/csv',\n",
    "        Body=csv_serializer.serialize(payload)\n",
    "    )\n",
    "    variant_name = resp['ResponseMetadata']['HTTPHeaders']['x-amzn-invoked-production-variant']\n",
    "    return float(resp['Body'].read().decode('utf-8').strip()), variant_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def approval(token, result):\n",
    "    if token is None:\n",
    "        return\n",
    "    \n",
    "    codepipeline.put_approval_result(\n",
    "      pipelineName=pipeline_name,\n",
    "      stageName='DeployApproval',\n",
    "      actionName='ApproveDeploy',\n",
    "      result=result,\n",
    "      token=token\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def approve(b):\n",
    "    result={\n",
    "        'summary': 'This is a great model! Put into production.',\n",
    "        'status': 'Approved'\n",
    "    }\n",
    "    approval(get_approval_token(), result) \n",
    "    button_box.close()\n",
    "    start_monitoring()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def reject(b):\n",
    "    result={\n",
    "        'summary': 'This is a rubbish model. Discard it',\n",
    "        'status': 'Rejected'\n",
    "    }\n",
    "    approval(get_approval_token(), result)\n",
    "    button_box.close()\n",
    "    start_monitoring()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def start_monitoring():\n",
    "    global button_box\n",
    "    \n",
    "    running = True\n",
    "    while running:\n",
    "        steps_ok = 0\n",
    "        for k,action in enumerate(get_actions()):\n",
    "            if action['actionStatus'] == 'Failed':\n",
    "                bar.bar_style='danger'\n",
    "                label.value='Ops! Something went wrong Stage[{}] Action[{}]'.format(\n",
    "                    action['stageName'], action['actionName'])\n",
    "                running = False\n",
    "                return\n",
    "\n",
    "            elif action['actionStatus'] == 'InProgress':\n",
    "                if get_approval_token() is not None:\n",
    "                    display(button_box)\n",
    "                    running = False\n",
    "                break\n",
    "            elif action['actionStatus'] == 'Old':\n",
    "                break\n",
    "            elif action['actionStatus'] == 'Succeeded':\n",
    "                steps_ok += 1\n",
    "        \n",
    "        label.value = \"Actions {}/{} - Current: Stage[{}] Action[{}]\".format( \n",
    "                k+1,max_actions, action['stageName'], action['actionName'] )\n",
    "        bar.value = steps_ok\n",
    "\n",
    "        if steps_ok == max_actions:\n",
    "            running = False\n",
    "        else:    \n",
    "            time.sleep(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Job monitoring"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "codepipeline = boto3.client('codepipeline')\n",
    "sm = boto3.client('sagemaker-runtime')\n",
    "\n",
    "model_prefix='iris-model'\n",
    "pipeline_name = 'iris-model-pipeline'\n",
    "endpoint_name_mask='{}-%s'.format(model_prefix)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "approve_btn = widgets.Button(description=\"Approve\", button_style='success', icon='check')\n",
    "reject_btn = widgets.Button(description=\"Reject\", button_style='danger', icon='close')\n",
    "approve_btn.on_click(approve)\n",
    "reject_btn.on_click(reject)\n",
    "button_box = widgets.HBox([approve_btn, reject_btn])\n",
    "                \n",
    "max_actions = len(get_actions())\n",
    "label = widgets.Label(value=\"Loading...\")\n",
    "bar = widgets.IntProgress( value=0, min=0, max=max_actions, step=1, bar_style='info' )\n",
    "info_box = widgets.VBox([label, bar])\n",
    "\n",
    "display(info_box)\n",
    "start_monitoring()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Now, if everything went fine, we can test our models"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "payload = [4.6, 3.1, 1.5, 0.2]\n",
    "\n",
    "print( \"DSV\")\n",
    "print( \"Classifier: %s, Variant Name: %s\" % test_endpoint( endpoint_name_mask % ('development'), payload ) )\n",
    "\n",
    "print( \"\\nPRD\")\n",
    "print( \"Classifier: %s, Variant Name: %s\" % test_endpoint( endpoint_name_mask % ('production'), payload ) )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}