air_force_model_server.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run with:\n",
    "#  $  FLASK_ENV=development FLASK_APP=air_force_model_server.py flask run\n",
    "#\n",
    "# Note.  This will run the server twice, as a feature of the reloader.\n",
    "#\n",
    "from flask import Flask, jsonify, request\n",
    "import numpy as np\n",
    "import math\n",
    "import pandas as pd\n",
    "import io\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.nn.functional as F"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# For reusing the same categorical encoder from training\n",
    "#\n",
    "import category_encoders as ce\n",
    "import pickle"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Variables and functions shared with training\n",
    "#\n",
    "import air_force_shared as afs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# For reading the model via URL\n",
    "#\n",
    "import os\n",
    "import sys\n",
    "from urllib.request import urlopen\n",
    "from shutil import copyfileobj\n",
    "from tempfile import NamedTemporaryFile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "# For reading command line\n",
    "#\n",
    "import argparse"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "# This version will load the model from a file passed in from the environment,\n",
    "# or from the current directory if the environment variable is missing.\n",
    "#\n",
    "# Loading from the environment will scale better during production,\n",
    "# as we can point the server to any model by just changing the environment.\n",
    "#\n",
    "def load_model_from_file():\n",
    "    m = afs.get_model()\n",
    "    model_params_location = afs.get_file_location(afs.MODEL_FILE)\n",
    "\n",
    "    # Note the use of the CPU here.  Currently, the docker image does not\n",
    "    # support CUDA. FIXME: It should be possible to do a GPU at runtime.\n",
    "    #\n",
    "    m.load_state_dict(torch.load(model_params_location, map_location='cpu'))\n",
    "    return m\n",
    "\n",
    "# This model loader will fetch the model from a URL in the environment.\n",
    "# Used if this server is to be hosted on a cloud service.\n",
    "#\n",
    "def load_model_from_url():\n",
    "    m = afs.get_model()\n",
    "    model_params_url = afs.get_url_location(afs.MODEL_FILE)\n",
    "    with urlopen(url) as fsrc, NameTemporaryFile() as fdst:\n",
    "        copyfileobj(fsrc, fdst)\n",
    "        # Note the use of the CPU here.  Currently, the docker image does not\n",
    "        # support CUDA. FIXME: It should be possible to do a GPU at runtime.\n",
    "        #\n",
    "        m.load_state_dict(torch.load(fdst, map_location='cpu'))\n",
    "    return m\n",
    "\n",
    "# Load the model.  If there is a missing environment variable,\n",
    "# or the model couldn't be loaded for any reason,\n",
    "# just stop the server entirely.\n",
    "#\n",
    "model = load_model_from_file()\n",
    "print('Loaded model:', model)\n",
    "if model == None:\n",
    "    sys.exit()\n",
    "\n",
    "# Set the model's state to 'eval()' so it won't run\n",
    "# backprop or use dropout layers (they are only used during training).\n",
    "#\n",
    "model.eval()\n",
    "\n",
    "# Transform input data into a tensor, to be used for inference.\n",
    "#\n",
    "# Note: We are assuming input is a table, not a single row.\n",
    "# \n",
    "# Similar to the function on the trainer, but does not include labels.\n",
    "#\n",
    "def transform_input(data, cat_features, cont_features, norm_prefix):\n",
    "\n",
    "    # This will fail if any of the continuous variables are missing.\n",
    "    #\n",
    "    features = (cat_features + cont_features)\n",
    "    feature_data = data[features].copy()\n",
    "\n",
    "    # Deal with nulls on input.\n",
    "    #\n",
    "    empty_string_replaced = feature_data.replace('', np.nan, inplace=False)\n",
    "    data_without_nulls = empty_string_replaced.dropna()\n",
    "    data_without_nulls.reset_index(drop=True, inplace=True)\n",
    "\n",
    "    # Convert the categorical feature values into one-hot vectors, removing\n",
    "    # the original categorical feature columns.  Reuse the same categorical\n",
    "    # encoder that we trained on.\n",
    "    #\n",
    "    cat_enc = afs.read_object(afs.get_file_location(afs.CATEGORICAL_ENCODER_FILE))\n",
    "    with_one_hot = cat_enc.transform(data_without_nulls.copy())\n",
    "\n",
    "    # The continuous encoder was written by us, so its methods aren't\n",
    "    # exactly consistent with the categorical data encoder.\n",
    "    #\n",
    "    cont_enc = afs.read_object(afs.get_file_location(afs.CONTINUOUS_ENCODER_FILE))\n",
    "\n",
    "    # Convert the continuous feature values into normalized features,\n",
    "    # removing the original continuous feature columns.\n",
    "    #\n",
    "    with_normalized = afs.fit_normalized(with_one_hot, cont_enc, norm_prefix)\n",
    "\n",
    "    encoded_list = with_normalized.values.tolist()  # to python list\n",
    "\n",
    "    return torch.FloatTensor(encoded_list)  # to tensor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The output is a vector of log_softmax (that is, a vector\n",
    "# of log-probabilities over the class).  We'll convert \n",
    "# these to just a single prediction for each label.\n",
    "#\n",
    "# (This is exactly the kind of calculation that is often\n",
    "# glossed over in models, where the modeler assumes that\n",
    "# the reader somehow knows by looking at an obscure function\n",
    "# what it's supposed to represent.)\n",
    "#\n",
    "# The shape of both of the results of topk() is [batch_size, 1],\n",
    "# so we flatten down to just [batch_size], so the result\n",
    "# can be compared against the labels.\n",
    "#\n",
    "def get_probabilities(model_prediction):\n",
    "    return torch.exp(model_prediction)            \n",
    "\n",
    "# Return the probabilities for each label prediction\n",
    "#\n",
    "def get_each_label_probability(model_prediction):\n",
    "    predicted_probs = get_probabilities(model_prediction)            \n",
    "    _, top_class = predicted_probs.topk(1, dim=1)\n",
    "\n",
    "    return torch.flatten(top_class)\n",
    "\n",
    "# Return the index within a vector of labels that indicates the prediction\n",
    "#\n",
    "def get_prediction_index(model_prediction):\n",
    "    predicted_labels = get_each_label_probability(model_prediction)\n",
    "    return predicted_labels.item()\n",
    "\n",
    "# Return the human-readable (textual) prediction\n",
    "#\n",
    "def get_human_prediction(model_prediction, collection):\n",
    "    prediction_index = get_prediction_index(model_prediction)\n",
    "    return collection['label_outputs'][prediction_index]\n",
    "\n",
    "def get_json_response(input_tensor, collection):\n",
    "    model_prediction = model(input_tensor)\n",
    "    probs = get_probabilities(model_prediction).detach().numpy()[0].tolist()\n",
    "    return jsonify({\n",
    "        'predict_index': get_prediction_index(model_prediction),\n",
    "        'predict_human': get_human_prediction(model_prediction, collection),\n",
    "        'probabilities': probs\n",
    "    })\n",
    "\n",
    "# This is the entrypoint when deploying.\n",
    "# See run-model-service.sh.\n",
    "#\n",
    "def create_app():\n",
    "    app = Flask(__name__)\n",
    "\n",
    "    # Get the configuration for datastore and model.\n",
    "    #\n",
    "    config = afs.read_object(afs.get_file_location(afs.CONFIG_FILE))\n",
    "    print(\"Inference service was started.\")\n",
    "\n",
    "    @app.route(\"/\")\n",
    "    def status():\n",
    "        return jsonify({\"status\": \"ok\"})\n",
    "\n",
    "    @app.route('/gethyper', methods=['GET', 'POST'])\n",
    "    def gethyper():\n",
    "        hyper = afs.read_object(afs.get_file_location(afs.HYPER_FILE))\n",
    "        return jsonify({'hyper': hyper})\n",
    "\n",
    "    @app.route('/getmodel', methods=['GET', 'POST'])\n",
    "    def getmodel():\n",
    "        # This is the function that print() uses.\n",
    "        #\n",
    "        modelrep = model.__repr__()\n",
    "        return jsonify({'model': modelrep})\n",
    "\n",
    "    # The prediction stub just generates a zero tensor to\n",
    "    # test the model\n",
    "    #\n",
    "    @app.route('/predict_zeroes', methods=['GET', 'POST'])\n",
    "    def predict_zeroes():\n",
    "        hyper = afs.read_object(afs.get_file_location(afs.HYPER_FILE))\n",
    "        input_tensor = torch.zeros(1, hyper.input_size)  # one row of columns with zeroes\n",
    "        return get_json_response(input_tensor, config['collection'])\n",
    "\n",
    "    @app.route('/predict_ones', methods=['GET', 'POST'])\n",
    "    def predict_ones():\n",
    "        hyper = afs.read_object(afs.get_file_location(afs.HYPER_FILE))\n",
    "        input_tensor = torch.ones(1, hyper.input_size)  # one row of columns with ones\n",
    "        return get_json_response(input_tensor, config['collection'])\n",
    "\n",
    "    @app.route('/predict_rand', methods=['GET', 'POST'])\n",
    "    def predict_rand():\n",
    "        hyper = afs.read_object(afs.get_file_location(afs.HYPER_FILE))\n",
    "        input_tensor = torch.rand(1, hyper.input_size)  # one row of columns with random values\n",
    "        return get_json_response(input_tensor, config['collection'])\n",
    "\n",
    "    # Allow the POST input to be either an array of samples, or a single row.\n",
    "    # Currently, only works for one row.  FIXME.\n",
    "    #\n",
    "    @app.route('/predict', methods=['GET', 'POST'])\n",
    "    def predict():\n",
    "        content = request.get_json()\n",
    "        if content != None:\n",
    "            list_content = content if type(content) is list else [content]\n",
    "\n",
    "            # The Docker image may throw a deprecation warning, like this:\n",
    "            #   ... pandas.io.json.json_normalize is deprecated,\n",
    "            #       use pandas.json_normalize instead\n",
    "            # However, don't make that change until the non-docker environment\n",
    "            # uses the same pandas version as the docker image.\n",
    "            #\n",
    "            raw_data = pd.io.json.json_normalize(list_content)\n",
    "            input_tensor = transform_input(raw_data,\n",
    "                config['collection']['cat_features'],\n",
    "                config['collection']['cont_features'], afs.norm_prefix)\n",
    "            return get_json_response(input_tensor, config['collection'])\n",
    "        else:\n",
    "            return jsonify({\n",
    "                'predict_index': None,\n",
    "                'predict_human': None,\n",
    "                'probabilities': None\n",
    "            })\n",
    "\n",
    "    return app"
   ]
  }
 ],
 "metadata": {
  "jupytext": {
   "cell_metadata_filter": "-all",
   "main_language": "python",
   "notebook_metadata_filter": "-all"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}