activity_infer.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np \n",
    "import os\n",
    "import glob\n",
    "import cv2\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "from mediapipe import solutions\n",
    "from mediapipe.framework.formats import landmark_pb2\n",
    "import numpy as np\n",
    "\n",
    "import mediapipe as mp\n",
    "from mediapipe.tasks import python\n",
    "from mediapipe.tasks.python import vision\n",
    "import glob\n",
    "from tqdm import tqdm\n",
    "from collections import OrderedDict, defaultdict\n",
    "from IPython.display import clear_output\n",
    "import time\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "device(type='cuda')"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')\n",
    "device"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "class LSTM(nn.Module):\n",
    "    \n",
    "    def __init__(self,input_dim,hidden_dim,output_dim,layer_num, seq_len):\n",
    "        super(LSTM,self).__init__()\n",
    "        self.hidden_dim = hidden_dim\n",
    "        self.output_dim = output_dim\n",
    "        self.lstm = torch.nn.LSTM(input_dim,hidden_dim,layer_num,batch_first=True)\n",
    "        self.fc = torch.nn.Linear(hidden_dim,output_dim)\n",
    "        self.bn = nn.BatchNorm1d(seq_len)\n",
    "        \n",
    "    def forward(self,inputs):\n",
    "        x = self.bn(inputs)\n",
    "        lstm_out,(hn,cn) = self.lstm(x)\n",
    "        out = self.fc(lstm_out[:,-1,:])\n",
    "        return out"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "LSTM(\n",
       "  (lstm): LSTM(50, 128, num_layers=3, batch_first=True)\n",
       "  (fc): Linear(in_features=128, out_features=2, bias=True)\n",
       "  (bn): BatchNorm1d(30, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)\n",
       ")"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "n_hidden = 128\n",
    "n_joints = 25*2\n",
    "n_categories = 2\n",
    "n_layer = 3\n",
    "seq_len = 30\n",
    "model = LSTM(n_joints,n_hidden,n_categories,n_layer, seq_len).to(device)\n",
    "\n",
    "\n",
    "model.load_state_dict(torch.load('rnn_train_1_dec7.pth'))\n",
    "model.eval()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_imgs(video_fn, skip_frames=1):\n",
    "    video=cv2.VideoCapture(video_fn)\n",
    "    imgs=[]\n",
    "    frame_count=0\n",
    "    while(video.isOpened()):\n",
    "        ret, frame = video.read()\n",
    "        if ret==False: break\n",
    "        if frame_count%skip_frames==0:\n",
    "            imgs.append(frame)\n",
    "        frame_count+=1\n",
    "    video.release() \n",
    "    return imgs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO: Created TensorFlow Lite XNNPACK delegate for CPU.\n"
     ]
    }
   ],
   "source": [
    "pose = mp.solutions.pose.Pose(static_image_mode=False, min_detection_confidence=0.5, min_tracking_confidence=0.5)\n",
    "def extract_keypoints(image_rgb): \n",
    "    try:\n",
    "        results = pose.process(image_rgb)\n",
    "        landmarks=results.pose_landmarks.landmark\n",
    "    except Exception as e:\n",
    "        # print('Error file=', fn)\n",
    "        # print('Error=', e)\n",
    "        return None\n",
    "    xys=[]\n",
    "    for landmark in landmarks:\n",
    "        xys.append([landmark.x, landmark.y])\n",
    "    xys=np.array(xys)\n",
    "    return xys"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "dir=\"/home/ns/data_activity/clips_eating/\"\n",
    "fn=dir+\"Dain Eating 1_1.mp4\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "52"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "imgs=extract_imgs(fn, skip_frames=1)\n",
    "len(imgs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(52, 50)"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kpss=[]\n",
    "for img in imgs:\n",
    "    kps=extract_keypoints(img)\n",
    "    kpss.append(kps[:25].flatten())\n",
    "\n",
    "kpss=np.array(kpss)\n",
    "kpss.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "torch.Size([1, 30, 50])"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "Xt=kpss[:seq_len]\n",
    "Xt=torch.tensor(Xt).unsqueeze(0).float().to(device)\n",
    "Xt.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "logits=model(Xt).detach().cpu().numpy()\n",
    "po=np.argmax(logits)\n",
    "po"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}