doc/Getting Started.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "219c0a05",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Connecting root@127.0.0.1:3306\n"
     ]
    }
   ],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import json\n",
    "import datetime\n",
    "import io\n",
    "import os\n",
    "import sys\n",
    "\n",
    "from IPython.display import Video as JupyterVideo\n",
    "from IPython.display import HTML\n",
    "from IPython.display import display\n",
    "from os import system, name\n",
    "\n",
    "os.environ['PYOPENGL_PLATFORM'] = 'osmesa'\n",
    "\n",
    "import pose_pipeline\n",
    "from pose_pipeline import *\n",
    "from pose_pipeline.utils.jupyter import play, play_grid\n",
    "pose_pipeline.set_environmental_variables()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "496b51da",
   "metadata": {},
   "source": [
    "![Data organization](erd.png)\n",
    "\n",
    "## Import test videos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bc572807",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pose_pipeline.utils.video_format import insert_local_video\n",
    "\n",
    "test_videos_path = 'PLACE/PATH/HERE'\n",
    "files = os.listdir(test_videos_path)\n",
    "\n",
    "for f in files:\n",
    "    insert_local_video(f, datetime.now(), os.path.join(test_videos_path, f), video_project='DEMO_TEST')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6fa956fd",
   "metadata": {},
   "outputs": [],
   "source": [
    "# List videos in database\n",
    "VideoInfo.populate(filt)\n",
    "Video & 'video_project=\"DEMO_TEST\"'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "64d6fce1",
   "metadata": {},
   "source": [
    "## Run tracking algorithm\n",
    "\n",
    "After running this, use the GUI to annotate the video to indicate the subject of interest."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8b94e962",
   "metadata": {},
   "outputs": [],
   "source": [
    "# select tracking method\n",
    "\n",
    "filt = 'video_project=\"DEMO_TEST\"'\n",
    "\n",
    "keys = ((Video & filt) - TrackingBboxMethod).fetch('KEY')\n",
    "\n",
    "print(f'Missing tracking methods: {len(keys)}')\n",
    "tracking_method = (TrackingBboxMethodLookup & 'tracking_method_name=\"MMTrack_deepsort\"').fetch1('tracking_method')\n",
    "for key in keys:\n",
    "    key['tracking_method'] = tracking_method\n",
    "    TrackingBboxMethod.insert1(key)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "df3ebc1a",
   "metadata": {},
   "outputs": [],
   "source": [
    "OpenPose.populate(filt)\n",
    "TrackingBbox.populate(filt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "70b426df",
   "metadata": {},
   "outputs": [],
   "source": [
    "# produce visualizations of these outputs used for annotation\n",
    "\n",
    "BlurredVideo.populate(filt)\n",
    "TrackingBboxVideo.populate(filt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "daafd639",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run annotation GUI\n",
    "to_label = ((TrackingBboxVideo & filt) - PersonBboxValid).fetch('KEY')\n",
    "\n",
    "def assign_video(labeling_key, valid, tracks=None):\n",
    "    \"\"\"\n",
    "    Params:\n",
    "        labeling_key: dictionary indicating the video being visualized to annotate\n",
    "        valid: boolean indicating a valid person could be identified\n",
    "        tracks: list matching the numbers associated with the person in video\n",
    "    \"\"\"\n",
    "    \n",
    "    # video_subject_id = 0 corresponds to a valid subject, and\n",
    "    # video_subject_id = -1 corresponds to a video with bad person detection\n",
    "    # the keep_tracks fields should correspond to the number(s) overlying the\n",
    "    # person in the video\n",
    "\n",
    "    video_key = (TrackingBbox & labeling_key).fetch1('KEY')\n",
    "\n",
    "    key = video_key.copy()\n",
    "    if valid:\n",
    "        key['video_subject_id'] = 0\n",
    "        key['keep_tracks'] = tracks\n",
    "    else:\n",
    "        key['video_subject_id'] = -1\n",
    "        key['keep_tracks'] = []\n",
    "        \n",
    "    print(f'Inserting {key} into database') #Inserts into database \n",
    "    PersonBboxValid.insert1(key)\n",
    "    \n",
    "def unlabel(labeling_key):\n",
    "    (PersonBboxValid & labeling_key).delete()\n",
    "\n",
    "\n",
    "def show_annotation(idx):\n",
    "    \n",
    "    labeling_key = to_label[idx]\n",
    "    print(f'Showing: {labeling_key}')\n",
    "    \n",
    "    video = (TrackingBboxVideo & labeling_key).fetch1('output_video')\n",
    "    \n",
    "    # get the track IDs present in the video\n",
    "    tracks = (TrackingBbox & labeling_key).fetch1('tracks')\n",
    "    tracks = np.unique([[t['track_id'] for track in tracks for t in track]]).tolist()\n",
    "    \n",
    "    display(JupyterVideo(video, height=480, html_attributes=\"controls muted autoplay\"))\n",
    "    practice = widgets.ToggleButtons(\n",
    "        options= ['Default', *tracks, 'Multiple', 'Skip','Undo', 'Invalid'],\n",
    "        disabled=False,\n",
    "        button_style='info', # 'success', 'info', 'warning', 'danger' or ''\n",
    "        tooltips=[\"Can't Identify Subject\", '0', '1', '2', '3', 'Skip']\n",
    "    )   \n",
    "    display(practice)\n",
    "    \n",
    "    multiple = widgets.Text(\n",
    "        value='',\n",
    "        placeholder='',\n",
    "        description='Multiple',\n",
    "        disabled=False\n",
    "    )\n",
    "    display(multiple)\n",
    "\n",
    "    def on_click(change):\n",
    "        value = change['new']\n",
    "        if value == 'Multiple':\n",
    "            value = [int(v) for v in multiple.value.split(',')]\n",
    "            assign_video(labeling_key, True, value)\n",
    "        elif value == 'Skip':\n",
    "            print('Skipping')\n",
    "        elif value == 'Invalid':\n",
    "            print('Flagging Invalid')\n",
    "            assign_video(labeling_key, False)\n",
    "        elif value == 'Undo':\n",
    "            unlabel(to_label[idx-1])\n",
    "            show_annotation(idx-1)\n",
    "            return\n",
    "        else:\n",
    "            value = int(value)\n",
    "            assign_video(labeling_key, True, [value])\n",
    "       \n",
    "        os.remove(video)\n",
    "        \n",
    "        show_annotation(idx + 1)\n",
    "\n",
    "    practice.observe(on_click, 'value'\"\")\n",
    "    \"\"\n",
    "show_annotation(0)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4221a864",
   "metadata": {},
   "source": [
    "## Perform additional analysis on the selected subject"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d2703b77",
   "metadata": {},
   "outputs": [],
   "source": [
    "# compute the final bounding box for subjects of interest\n",
    "pose_pipeline.PersonBbox.populate(filt)\n",
    "\n",
    "# select the top down approach to run\n",
    "top_down_method = (pose_pipeline.TopDownMethodLookup & 'top_down_method_name=\"MMPose\"').fetch1('top_down_method')\n",
    "keys = ((pose_pipeline.PersonBbox & filt) - (pose_pipeline.TopDownMethod & {'smpl_method': smpl_method})).fetch('KEY')\n",
    "print(f'Missing top down methods: {len(keys)}')\n",
    "for key in keys:\n",
    "    key['top_down_method'] = top_down_method\n",
    "    pose_pipeline.TopDownMethod.insert1(key)\n",
    "\n",
    "# select an SMPL method to run\n",
    "smpl_method = (pose_pipeline.SMPLMethodLookup & 'smpl_method_name=\"PARE\"').fetch1('smpl_method')\n",
    "keys = ((pose_pipeline.PersonBbox & filt) - (pose_pipeline.SMPLMethod & {'smpl_method': smpl_method})).fetch('KEY')\n",
    "print(f'Missing SMPL methods: {len(keys)}')\n",
    "for key in keys:\n",
    "    key['smpl_method'] = smpl_method\n",
    "    pose_pipeline.SMPLMethod.insert1(key)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "96b0d3b6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# populate the selected analyses\n",
    "\n",
    "TopDownPerson.populate(filt)\n",
    "SMPLPerson.populate(filt)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0eee8339",
   "metadata": {},
   "source": [
    "## Produce and review visualizations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9319194a",
   "metadata": {},
   "outputs": [],
   "source": [
    "TopDownPersonVideo.populate(filt)\n",
    "SMPLPersonVideo.populate(filt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e797e258",
   "metadata": {},
   "outputs": [],
   "source": [
    "keys = (TopDownPersonVideo & filt).fetch('KEY')\n",
    "play(TopDownPersonVideo & keys[0])"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}