starter.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Deploy and monitor a machine learning workflow for Image Classification"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setting up this notebook\n",
    "\n",
    "Notes about the instance size and kernel setup: this notebook has been tested on\n",
    "\n",
    "1. The `Python 3 (Data Science)` kernel\n",
    "2. The `ml.t3.medium` Sagemaker notebook instance\n",
    "\n",
    "## Data Staging\n",
    "\n",
    "We'll use a sample dataset called CIFAR to simulate the challenges Scones Unlimited are facing in Image Classification. In order to start working with CIFAR we'll need to:\n",
    "\n",
    "1. Extract the data from a hosting service\n",
    "2. Transform it into a usable shape and format\n",
    "3. Load it into a production system\n",
    "\n",
    "In other words, we're going to do some simple ETL!\n",
    "\n",
    "### 1. Extract the data from the hosting service\n",
    "\n",
    "In the cell below, define a function `extract_cifar_data` that extracts python version of the CIFAR-100 dataset. The CIFAR dataaset is open source and generously hosted by the University of Toronto at: https://www.cs.toronto.edu/~kriz/cifar-100-python.tar.gz "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "\n",
    "def extract_cifar_data(url, filename=\"cifar.tar.gz\"):\n",
    "    \"\"\"A function for extracting the CIFAR-100 dataset and storing it as a gzipped file\n",
    "    \n",
    "    Arguments:\n",
    "    url      -- the URL where the dataset is hosted\n",
    "    filename -- the full path where the dataset will be written\n",
    "    \n",
    "    \"\"\"\n",
    "    \n",
    "    # Todo: request the data from the data url\n",
    "    # Hint: use `requests.get` method\n",
    "    r = \n",
    "    with open(filename, \"wb\") as file_context:\n",
    "        file_context.write(r.content)\n",
    "    return"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's test it out! Run the following cell and check whether a new file `cifar.tar.gz` is created in the file explorer."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "extract_cifar_data(\"https://www.cs.toronto.edu/~kriz/cifar-100-python.tar.gz\")     "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Transform the data into a usable shape and format\n",
    "\n",
    "Clearly, distributing the data as a gzipped archive makes sense for the hosting service! It saves on bandwidth, storage, and it's a widely-used archive format. In fact, it's so widely used that the Python community ships a utility for working with them, `tarfile`, as part of its Standard Library. Execute the following cell to decompress your extracted dataset:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import tarfile\n",
    "\n",
    "with tarfile.open(\"cifar.tar.gz\", \"r:gz\") as tar:\n",
    "    tar.extractall()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A new folder `cifar-100-python` should be created, containing `meta`, `test`, and `train` files. These files are `pickles` and the [CIFAR homepage](https://www.cs.toronto.edu/~kriz/cifar.html) provides a simple script that can be used to load them. We've adapted the script below for you to run:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pickle\n",
    "\n",
    "with open(\"./cifar-100-python/meta\", \"rb\") as f:\n",
    "    dataset_meta = pickle.load(f, encoding='bytes')\n",
    "\n",
    "with open(\"./cifar-100-python/test\", \"rb\") as f:\n",
    "    dataset_test = pickle.load(f, encoding='bytes')\n",
    "\n",
    "with open(\"./cifar-100-python/train\", \"rb\") as f:\n",
    "    dataset_train = pickle.load(f, encoding='bytes')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Feel free to explore the datasets\n",
    "\n",
    "dataset_train.keys()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As documented on the homepage, `b'data'` contains rows of 3073 unsigned integers, representing three channels (red, green, and blue) for one 32x32 pixel image per row."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "32*32*3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For a simple gut-check, let's transform one of our images. Each 1024 items in a row is a channel (red, green, then blue). Each 32 items in the channel are a row in the 32x32 image. Using python, we can stack these channels into a 32x32x3 array, and save it as a PNG file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "# Each 1024 in a row is a channel (red, green, then blue)\n",
    "row = dataset_train[b'data'][0]\n",
    "red, green, blue = row[0:1024], row[1024:2048], row[2048:]\n",
    "\n",
    "# Each 32 items in the channel are a row in the 32x32 image\n",
    "red = red.reshape(32,32)\n",
    "green = green.reshape(32,32)\n",
    "blue = blue.reshape(32,32)\n",
    "\n",
    "# Combine the channels into a 32x32x3 image!\n",
    "combined = np.dstack((red,green,blue))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For a more concise version, consider the following:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# All in one:\n",
    "test_image = np.dstack((\n",
    "    row[0:1024].reshape(32,32),\n",
    "    row[1024:2048].reshape(32,32),\n",
    "    row[2048:].reshape(32,32)\n",
    "))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "plt.imshow(test_image);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Looks like a cow! Let's check the label. `dataset_meta` contains label names in order, and `dataset_train` has a list of labels for each row."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset_train[b'fine_labels'][0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our image has a label of `19`, so let's see what the 19th item is in the list of label names."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(dataset_meta[b'fine_label_names'][19])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ok! 'cattle' sounds about right. By the way, using the previous two lines we can do:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n = 0\n",
    "print(dataset_meta[b'fine_label_names'][dataset_train[b'fine_labels'][n]])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we know how to check labels, is there a way that we can also check file names? `dataset_train` also contains a `b'filenames'` key. Let's see what we have here:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(dataset_train[b'filenames'][0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\"Taurus\" is the name of a subspecies of cattle, so this looks like a pretty reasonable filename. To save an image we can also do:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.imsave(\"file.png\", test_image)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Your new PNG file should now appear in the file explorer -- go ahead and pop it open to see!\n",
    "\n",
    "Now that you know how to reshape the images, save them as files, and capture their filenames and labels, let's just capture all the bicycles and motorcycles and save them. Scones Unlimited can use a model that tells these apart to route delivery drivers automatically.\n",
    "\n",
    "In the following cell, identify the label numbers for Bicycles and Motorcycles:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "# Todo: Filter the dataset_train and dataset_meta objects to find the label numbers for Bicycle and Motorcycles\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Good job! We only need objects with label 8 and 48 -- this drastically simplifies our handling of the data! Below we construct a dataframe for you, and you can safely drop the rows that don't contain observations about bicycles and motorcycles. Fill in the missing lines below to drop all other rows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Construct the dataframe\n",
    "df_train = pd.DataFrame({\n",
    "    \"filenames\": dataset_train[b'filenames'],\n",
    "    \"labels\": dataset_train[b'fine_labels'],\n",
    "    \"row\": range(len(dataset_train[b'filenames']))\n",
    "})\n",
    "\n",
    "# Drop all rows from df_train where label is not 8 or 48\n",
    "df_train = #TODO: Fill in\n",
    "\n",
    "# Decode df_train.filenames so they are regular strings\n",
    "df_train[\"filenames\"] = df_train[\"filenames\"].apply(\n",
    "    lambda x: x.decode(\"utf-8\")\n",
    ")\n",
    "\n",
    "\n",
    "df_test = pd.DataFrame({\n",
    "    \"filenames\": dataset_test[b'filenames'],\n",
    "    \"labels\": dataset_test[b'fine_labels'],\n",
    "    \"row\": range(len(dataset_test[b'filenames']))\n",
    "})\n",
    "\n",
    "# Drop all rows from df_test where label is not 8 or 48\n",
    "df_test = #TODO: Fill in\n",
    "\n",
    "# Decode df_test.filenames so they are regular strings\n",
    "df_test[\"filenames\"] = df_test[\"filenames\"].apply(\n",
    "    lambda x: x.decode(\"utf-8\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that the data is filtered for just our classes, we can save all our images."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!mkdir ./train\n",
    "!mkdir ./test"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the previous sections we introduced you to several key snippets of code:\n",
    "\n",
    "1. Grabbing the image data:\n",
    "\n",
    "```python\n",
    "dataset_train[b'data'][0]\n",
    "```\n",
    "\n",
    "2. A simple idiom for stacking the image data into the right shape\n",
    "\n",
    "```python\n",
    "import numpy as np\n",
    "np.dstack((\n",
    "    row[0:1024].reshape(32,32),\n",
    "    row[1024:2048].reshape(32,32),\n",
    "    row[2048:].reshape(32,32)\n",
    "))\n",
    "```\n",
    "\n",
    "3. A simple `matplotlib` utility for saving images\n",
    "\n",
    "```python\n",
    "plt.imsave(path+row['filenames'], target)\n",
    "```\n",
    "\n",
    "Compose these together into a function that saves all the images into the `./test` and `./train` directories. Use the comments in the body of the `save_images` function below to guide your construction of the function:\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_images():\n",
    "    #Grab the image data in row-major form\n",
    "    img = \n",
    "    \n",
    "    # Consolidated stacking/reshaping from earlier\n",
    "    target = \n",
    "    \n",
    "    # Save the image\n",
    "    \n",
    "    \n",
    "    # Return any signal data you want for debugging\n",
    "    return\n",
    "\n",
    "## TODO: save ALL images using the save_images function"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Load the data\n",
    "\n",
    "Now we can load the data into S3.\n",
    "\n",
    "Using the sagemaker SDK grab the current region, execution role, and bucket."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sagemaker\n",
    "\n",
    "\n",
    "bucket= ## TODO: fill in\n",
    "print(\"Default Bucket: {}\".format(bucket))\n",
    "\n",
    "region = ## TODO: fill in\n",
    "print(\"AWS Region: {}\".format(region))\n",
    "\n",
    "role = ## TODO: fill in\n",
    "print(\"RoleArn: {}\".format(role))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With this data we can easily sync your data up into S3!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "os.environ[\"DEFAULT_S3_BUCKET\"] = bucket\n",
    "!aws s3 sync ./train s3://${DEFAULT_S3_BUCKET}/train/\n",
    "!aws s3 sync ./test s3://${DEFAULT_S3_BUCKET}/test/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And that's it! You can check the bucket and verify that the items were uploaded.\n",
    "\n",
    "## Model Training\n",
    "\n",
    "For Image Classification, Sagemaker [also expects metadata](https://docs.aws.amazon.com/sagemaker/latest/dg/image-classification.html) e.g. in the form of TSV files with labels and filepaths. We can generate these using our Pandas DataFrames from earlier:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def to_metadata_file(df, prefix):\n",
    "    df[\"s3_path\"] = df[\"filenames\"]\n",
    "    df[\"labels\"] = df[\"labels\"].apply(lambda x: 0 if x==8 else 1)\n",
    "    return df[[\"row\", \"labels\", \"s3_path\"]].to_csv(\n",
    "        f\"{prefix}.lst\", sep=\"\\t\", index=False, header=False\n",
    "    )\n",
    "    \n",
    "to_metadata_file(df_train.copy(), \"train\")\n",
    "to_metadata_file(df_test.copy(), \"test\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also upload our manifest files:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "\n",
    "# Upload files\n",
    "boto3.Session().resource('s3').Bucket(\n",
    "    bucket).Object('train.lst').upload_file('./train.lst')\n",
    "boto3.Session().resource('s3').Bucket(\n",
    "    bucket).Object('test.lst').upload_file('./test.lst')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the `bucket` and `region` info we can get the latest prebuilt container to run our training job, and define an output location on our s3 bucket for the model. Use the `image_uris` function from the SageMaker SDK to retrieve the latest `image-classification` image below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Use the image_uris function to retrieve the latest 'image-classification' image \n",
    "algo_image = ## TODO: fill in\n",
    "s3_output_location = f\"s3://{bucket}/models/image_model\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We're ready to create an estimator! Create an estimator `img_classifier_model` that uses one instance of `ml.p3.2xlarge`. Ensure that y ou use the output location we defined above - we'll be referring to that later!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "img_classifier_model=sagemaker.estimator.Estimator(\n",
    "    ## TODO: define your estimator options    \n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also set a few key hyperparameters and define the inputs for our model:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "img_classifier_model.set_hyperparameters(\n",
    "    image_shape=, # TODO: Fill in\n",
    "    num_classes=, # TODO: Fill in\n",
    "    num_training_samples= # TODO: fill in\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `image-classification` image uses four input channels with very specific input parameters. For convenience, we've provided them below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.debugger import Rule, rule_configs\n",
    "from sagemaker.session import TrainingInput\n",
    "model_inputs = {\n",
    "        \"train\": sagemaker.inputs.TrainingInput(\n",
    "            s3_data=f\"s3://{bucket}/train/\",\n",
    "            content_type=\"application/x-image\"\n",
    "        ),\n",
    "        \"validation\": sagemaker.inputs.TrainingInput(\n",
    "            s3_data=f\"s3://{bucket}/test/\",\n",
    "            content_type=\"application/x-image\"\n",
    "        ),\n",
    "        \"train_lst\": sagemaker.inputs.TrainingInput(\n",
    "            s3_data=f\"s3://{bucket}/train.lst\",\n",
    "            content_type=\"application/x-image\"\n",
    "        ),\n",
    "        \"validation_lst\": sagemaker.inputs.TrainingInput(\n",
    "            s3_data=f\"s3://{bucket}/test.lst\",\n",
    "            content_type=\"application/x-image\"\n",
    "        )\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Great, now we can train the model using the model_inputs. In the cell below, call the `fit` method on our model,:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## TODO: train your model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If all goes well, you'll end up with a model topping out above `.8` validation accuracy. With only 1000 training samples in the CIFAR dataset, that's pretty good. We could definitely pursue data augmentation & gathering more samples to help us improve further, but for now let's proceed to deploy our model.\n",
    "\n",
    "### Getting ready to deploy\n",
    "\n",
    "To begin with, let's configure Model Monitor to track our deployment. We'll define a `DataCaptureConfig` below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.model_monitor import DataCaptureConfig\n",
    "\n",
    "data_capture_config = DataCaptureConfig(\n",
    "    ## TODO: Set config options\n",
    "    destination_s3_uri=f\"s3://{bucket}/data_capture\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note the `destination_s3_uri` parameter: At the end of the project, we can explore the `data_capture` directory in S3 to find crucial data about the inputs and outputs Model Monitor has observed on our model endpoint over time.\n",
    "\n",
    "With that done, deploy your model on a single `ml.m5.xlarge` instance with the data capture config attached:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "deployment = img_classifier_model.deploy(\n",
    "    ## TODO: fill in deployment options\n",
    "    data_capture_config=data_capture_config\n",
    "    )\n",
    "\n",
    "endpoint = deployment.endpoint_name\n",
    "print(endpoint)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note the endpoint name for later as well.\n",
    "\n",
    "Next, instantiate a Predictor:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictor = ## TODO: fill in"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the code snippet below we are going to prepare one of your saved images for prediction. Use the predictor to process the `payload`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.serializers import IdentitySerializer\n",
    "import base64\n",
    "\n",
    "predictor.serializer = IdentitySerializer(\"image/png\")\n",
    "with open(\"./test/bicycle_s_001789.png\", \"rb\") as f:\n",
    "    payload = f.read()\n",
    "\n",
    "    \n",
    "inference = ## TODO: Process the payload with your predictor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Your `inference` object is an array of two values, the predicted probability value for each of your classes (bicycle and motorcycle respectively.) So, for example, a value of `b'[0.91, 0.09]'` indicates the probability of being a bike is 91% and being a motorcycle is 9%."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(inference)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Draft Lambdas and Step Function Workflow\n",
    "\n",
    "Your operations team uses Step Functions to orchestrate serverless workflows. One of the nice things about Step Functions is that [workflows can call other workflows](https://docs.aws.amazon.com/step-functions/latest/dg/connect-stepfunctions.html), so the team can easily plug your workflow into the broader production architecture for Scones Unlimited.\n",
    "\n",
    "In this next stage you're going to write and deploy three Lambda functions, and then use the Step Functions visual editor to chain them together! Our functions are going to work with a simple data object:\n",
    "\n",
    "```python\n",
    "{\n",
    "    \"inferences\": [], # Output of predictor.predict\n",
    "    \"s3_key\": \"\", # Source data S3 key\n",
    "    \"s3_bucket\": \"\", # Source data S3 bucket\n",
    "    \"image_data\": \"\"  # base64 encoded string containing the image data\n",
    "}\n",
    "```\n",
    "\n",
    "A good test object that you can use for Lambda tests and Step Function executions, throughout the next section, might look like this:\n",
    "\n",
    "```python\n",
    "{\n",
    "  \"image_data\": \"\",\n",
    "  \"s3_bucket\": MY_BUCKET_NAME, # Fill in with your bucket\n",
    "  \"s3_key\": \"test/bicycle_s_000513.png\"\n",
    "}\n",
    "```\n",
    "\n",
    "Using these fields, your functions can read and write the necessary data to execute your workflow. Let's start with the first function. Your first Lambda function will copy an object from S3, base64 encode it, and then return it to the step function as `image_data` in an event.\n",
    "\n",
    "Go to the Lambda dashboard and create a new Lambda function with a descriptive name like \"serializeImageData\" and select thr 'Python 3.8' runtime. Add the same permissions as the SageMaker role you created earlier. (Reminder: you do this in the Configuration tab under \"Permissions\"). Once you're ready, use the starter code below to craft your Lambda handler:\n",
    "\n",
    "```python\n",
    "import json\n",
    "import boto3\n",
    "import base64\n",
    "\n",
    "s3 = boto3.client('s3')\n",
    "\n",
    "def lambda_handler(event, context):\n",
    "    \"\"\"A function to serialize target data from S3\"\"\"\n",
    "    \n",
    "    # Get the s3 address from the Step Function event input\n",
    "    key = ## TODO: fill in\n",
    "    bucket = ## TODO: fill in\n",
    "    \n",
    "    # Download the data from s3 to /tmp/image.png\n",
    "    ## TODO: fill in\n",
    "    \n",
    "    # We read the data from a file\n",
    "    with open(\"/tmp/image.png\", \"rb\") as f:\n",
    "        image_data = base64.b64encode(f.read())\n",
    "\n",
    "    # Pass the data back to the Step Function\n",
    "    print(\"Event:\", event.keys())\n",
    "    return {\n",
    "        'statusCode': 200,\n",
    "        'body': {\n",
    "            \"image_data\": image_data,\n",
    "            \"s3_bucket\": bucket,\n",
    "            \"s3_key\": key,\n",
    "            \"inferences\": []\n",
    "        }\n",
    "    }\n",
    "```\n",
    "\n",
    "The next function is responsible for the classification part - we're going to take the image output from the previous function, decode it, and then pass inferences back to the the Step Function.\n",
    "\n",
    "Because this Lambda will have runtime dependencies (i.e. the SageMaker SDK) you'll need to package them in your function. *Key reading:* https://docs.aws.amazon.com/lambda/latest/dg/python-package-create.html#python-package-create-with-dependency\n",
    "\n",
    "Create a new Lambda function with the same rights and a descriptive name, then fill in the starter code below for your classifier Lambda.\n",
    "\n",
    "```python\n",
    "import json\n",
    "import sagemaker\n",
    "import base64\n",
    "from sagemaker.serializers import IdentitySerializer\n",
    "\n",
    "# Fill this in with the name of your deployed model\n",
    "ENDPOINT = ## TODO: fill in\n",
    "\n",
    "def lambda_handler(event, context):\n",
    "\n",
    "    # Decode the image data\n",
    "    image = base64.b64decode(## TODO: fill in)\n",
    "\n",
    "    # Instantiate a Predictor\n",
    "    predictor = ## TODO: fill in\n",
    "\n",
    "    # For this model the IdentitySerializer needs to be \"image/png\"\n",
    "    predictor.serializer = IdentitySerializer(\"image/png\")\n",
    "    \n",
    "    # Make a prediction:\n",
    "    inferences = ## TODO: fill in\n",
    "    \n",
    "    # We return the data back to the Step Function    \n",
    "    event[\"inferences\"] = inferences.decode('utf-8')\n",
    "    return {\n",
    "        'statusCode': 200,\n",
    "        'body': json.dumps(event)\n",
    "    }\n",
    "```\n",
    "\n",
    "Finally, we need to filter low-confidence inferences. Define a threshold between 1.00 and 0.000 for your model: what is reasonble for you? If the model predicts at `.70` for it's highest confidence label, do we want to pass that inference along to downstream systems? Make one last Lambda function and tee up the same permissions:\n",
    "\n",
    "```python\n",
    "import json\n",
    "\n",
    "\n",
    "THRESHOLD = .93\n",
    "\n",
    "\n",
    "def lambda_handler(event, context):\n",
    "    \n",
    "    # Grab the inferences from the event\n",
    "    inferences = ## TODO: fill in\n",
    "    \n",
    "    # Check if any values in our inferences are above THRESHOLD\n",
    "    meets_threshold = ## TODO: fill in\n",
    "    \n",
    "    # If our threshold is met, pass our data back out of the\n",
    "    # Step Function, else, end the Step Function with an error\n",
    "    if meets_threshold:\n",
    "        pass\n",
    "    else:\n",
    "        raise(\"THRESHOLD_CONFIDENCE_NOT_MET\")\n",
    "\n",
    "    return {\n",
    "        'statusCode': 200,\n",
    "        'body': json.dumps(event)\n",
    "    }\n",
    "```\n",
    "Once you have tested the lambda functions, save the code for each lambda function in a python script called 'lambda.py'.\n",
    "\n",
    "With your lambdas in place, you can use the Step Functions visual editor to construct a workflow that chains them together. In the Step Functions console you'll have the option to author a Standard step function *Visually*.\n",
    "\n",
    "When the visual editor opens, you'll have many options to add transitions in your workflow. We're going to keep it simple and have just one: to invoke Lambda functions. Add three of them chained together. For each one, you'll be able to select the Lambda functions you just created in the proper order, filter inputs and outputs, and give them descriptive names.\n",
    "\n",
    "Make sure that you:\n",
    "\n",
    "1. Are properly filtering the inputs and outputs of your invokations (e.g. `$.body`)\n",
    "2. Take care to remove the error handling from the last function - it's supposed to \"fail loudly\" for your operations colleagues!\n",
    "\n",
    "Take a screenshot of your working step function in action and export the step function as JSON for your submission package."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "Great! Now you can use the files in `./test` as test files for our workflow. Depending on our threshold, our workflow should reliably pass predictions about images from `./test` on to downstream systems, while erroring out for inferences below our confidence threshold!\n",
    "\n",
    "### Testing and Evaluation\n",
    "\n",
    "Do several step function invokations using data from the `./test` folder. This process should give you confidence that the workflow both *succeeds* AND *fails* as expected. In addition, SageMaker Model Monitor will generate recordings of your data and inferences which we can visualize.\n",
    "\n",
    "Here's a function that can help you generate test inputs for your invokations:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "import boto3\n",
    "import json\n",
    "\n",
    "\n",
    "def generate_test_case():\n",
    "    # Setup s3 in boto3\n",
    "    s3 = boto3.resource('s3')\n",
    "    \n",
    "    # Randomly pick from sfn or test folders in our bucket\n",
    "    objects = s3.Bucket(bucket).objects.filter(\"test\")\n",
    "    \n",
    "    # Grab any random object key from that folder!\n",
    "    obj = random.choice([x.key for x in objects])\n",
    "    \n",
    "    return json.dumps({\n",
    "        \"image_data\": \"\",\n",
    "        \"s3_bucket\": bucket,\n",
    "        \"s3_key\": obj\n",
    "    })\n",
    "generate_test_case()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the Step Function dashboard for your new function, you can create new executions and copy in the generated test cases. Do several executions so that you can generate data you can evaluate and visualize.\n",
    "\n",
    "Once you've done several executions, let's visualize the record of our inferences. Pull in the JSONLines data from your inferences like so:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.s3 import S3Downloader\n",
    "\n",
    "# In S3 your data will be saved to a datetime-aware path\n",
    "# Find a path related to a datetime you're interested in\n",
    "data_path = ## TODO: fill in the path to your captured data\n",
    "\n",
    "S3Downloader.download(data_path, \"captured_data\")\n",
    "\n",
    "# Feel free to repeat this multiple times and pull in more data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The data are in JSONLines format, where multiple valid JSON objects are stacked on top of eachother in a single `jsonl` file. We'll import an open-source library, `jsonlines` that was purpose built for parsing this format."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install jsonlines\n",
    "import jsonlines"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can extract the data from each of the source files:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "# List the file names we downloaded\n",
    "file_handles = os.listdir(\"./captured_data\")\n",
    "\n",
    "# Dump all the data into an array\n",
    "json_data = []\n",
    "for jsonl in file_handles:\n",
    "    with jsonlines.open(f\"./captured_data/{jsonl}\") as f:\n",
    "        json_data.append(f.read())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The data should now be a list of dictionaries, with significant nesting. We'll give you an example of some code that grabs data out of the objects and visualizes it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define how we'll get our data\n",
    "def simple_getter(obj):\n",
    "    inferences = obj[\"captureData\"][\"endpointOutput\"][\"data\"]\n",
    "    timestamp = obj[\"eventMetadata\"][\"inferenceTime\"]\n",
    "    return json.loads(inferences), timestamp\n",
    "\n",
    "simple_getter(json_data[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, here's an example of a visualization you can build with this data. In this last part, you will take some time and build your own - the captured data has the input images, the resulting inferences, and the timestamps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Populate the data for the x and y axis\n",
    "x = []\n",
    "y = []\n",
    "for obj in json_data:\n",
    "    inference, timestamp = simple_getter(obj)\n",
    "    \n",
    "    y.append(max(inference))\n",
    "    x.append(timestamp)\n",
    "\n",
    "# Todo: here is an visualization example, take some time to build another visual that helps monitor the result\n",
    "# Plot the data\n",
    "plt.scatter(x, y, c=['r' if k<.94 else 'b' for k in y ])\n",
    "plt.axhline(y=0.94, color='g', linestyle='--')\n",
    "plt.ylim(bottom=.88)\n",
    "\n",
    "# Add labels\n",
    "plt.ylabel(\"Confidence\")\n",
    "plt.suptitle(\"Observed Recent Inferences\", size=14)\n",
    "plt.title(\"Pictured with confidence threshold for production use\", size=10)\n",
    "\n",
    "# Give it some pizzaz!\n",
    "plt.style.use(\"Solarize_Light2\")\n",
    "plt.gcf().autofmt_xdate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Todo: build your own visualization\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Congratulations!\n",
    "\n",
    "You've reached the end of the project. In this project you created an event-drivent ML workflow that can be incorporated into the Scones Unlimited production architecture. You used the SageMaker Estimator API to deploy your SageMaker Model and Endpoint, and you used AWS Lambda and Step Functions to orchestrate your ML workflow. Using SageMaker Model Monitor, you instrumented and observed your Endpoint, and at the end of the project you built a visualization to help stakeholders understand the performance of the Endpoint over time. If you're up for it, you can even go further with these stretch goals:\n",
    "\n",
    "* Extend your workflow to incorporate more classes: the CIFAR dataset includes other vehicles that Scones Unlimited can identify with this model.\n",
    "* Modify your event driven workflow: can you rewrite your Lambda functions so that the workflow can process multiple image inputs in parallel? Can the Step Function \"fan out\" to accomodate this new workflow?\n",
    "* Consider the test data generator we provided for you. Can we use it to create a \"dummy data\" generator, to simulate a continuous stream of input data? Or a big paralell load of data?\n",
    "* What if we want to get notified every time our step function errors out? Can we use the Step Functions visual editor in conjunction with a service like SNS to accomplish this? Try it out!\n",
    "\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (Data Science)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-east-1:081325390199:image/datascience-1.0"
  },
  "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.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}