{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import csv\nimport cv2\nimport itertools\nimport numpy as np\nimport pandas as pd\nimport os\nimport sys\nimport tempfile\nimport tqdm\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T11:56:12.089497Z","iopub.execute_input":"2023-07-28T11:56:12.089853Z","iopub.status.idle":"2023-07-28T11:56:23.836033Z","shell.execute_reply.started":"2023-07-28T11:56:12.089824Z","shell.execute_reply":"2023-07-28T11:56:23.834229Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Download model from TF Hub and check out inference code from GitHub\n!wget -q -O movenet_thunder.tflite https://tfhub.dev/google/lite-model/movenet/singlepose/thunder/tflite/float16/4?lite-format=tflite\n!git clone https://github.com/tensorflow/examples.git\npose_sample_rpi_path = os.path.join(os.getcwd(), 'examples/lite/examples/pose_estimation/raspberry_pi')\nsys.path.append(pose_sample_rpi_path)\n\n# Load MoveNet Thunder model\nimport utils\nfrom data import BodyPart\nfrom ml import Movenet\nmovenet = Movenet('movenet_thunder')\n\n# Define function to run pose estimation using MoveNet Thunder.\n# You'll apply MoveNet's cropping algorithm and run inference multiple times on\n# the input image to improve pose estimation accuracy.\ndef detect(input_tensor, inference_count=3):\n  \"\"\"Runs detection on an input image.\n\n  Args:\n    input_tensor: A [height, width, 3] Tensor of type tf.float32.\n      Note that height and width can be anything since the image will be\n      immediately resized according to the needs of the model within this\n      function.\n    inference_count: Number of times the model should run repeatly on the\n      same input image to improve detection accuracy.\n\n  Returns:\n    A Person entity detected by the MoveNet.SinglePose.\n  \"\"\"\n  image_height, image_width, channel = input_tensor.shape\n\n  # Detect pose using the full input image\n  movenet.detect(input_tensor.numpy(), reset_crop_region=True)\n\n  # Repeatedly using previous detection result to identify the region of\n  # interest and only croping that region to improve detection accuracy\n  for _ in range(inference_count - 1):\n    person = movenet.detect(input_tensor.numpy(), \n                            reset_crop_region=False)\n\n  return person\n","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2023-07-28T11:56:23.837724Z","iopub.execute_input":"2023-07-28T11:56:23.838341Z","iopub.status.idle":"2023-07-28T11:56:29.433089Z","shell.execute_reply.started":"2023-07-28T11:56:23.838314Z","shell.execute_reply":"2023-07-28T11:56:29.431156Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def draw_prediction_on_image(\n    image, person, crop_region=None, close_figure=True,\n    keep_input_size=False):\n  \"\"\"Draws the keypoint predictions on image.\n\n  Args:\n    image: An numpy array with shape [height, width, channel] representing the\n      pixel values of the input image.\n    person: A person entity returned from the MoveNet.SinglePose model.\n    close_figure: Whether to close the plt figure after the function returns.\n    keep_input_size: Whether to keep the size of the input image.\n\n  Returns:\n    An numpy array with shape [out_height, out_width, channel] representing the\n    image overlaid with keypoint predictions.\n  \"\"\"\n  # Draw the detection result on top of the image.\n  image_np = utils.visualize(image, [person])\n\n  # Plot the image with detection results.\n  height, width, channel = image.shape\n  aspect_ratio = float(width) / height\n  fig, ax = plt.subplots(figsize=(12 * aspect_ratio, 12))\n  im = ax.imshow(image_np)\n\n  if close_figure:\n    plt.close(fig)\n\n  if not keep_input_size:\n    image_np = utils.keep_aspect_ratio_resizer(image_np, (512, 512))\n\n  return image_np\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T11:56:29.435167Z","iopub.execute_input":"2023-07-28T11:56:29.435513Z","iopub.status.idle":"2023-07-28T11:56:29.443557Z","shell.execute_reply.started":"2023-07-28T11:56:29.435480Z","shell.execute_reply":"2023-07-28T11:56:29.442180Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class MoveNetPreprocessor(object):\n  \"\"\"Helper class to preprocess pose sample images for classification.\"\"\"\n\n  def __init__(self,\n               images_in_folder,\n               images_out_folder,\n               csvs_out_path):\n    \"\"\"Creates a preprocessor to detection pose from images and save as CSV.\n\n    Args:\n      images_in_folder: Path to the folder with the input images. It should\n        follow this structure:\n        yoga_poses\n        |__ downdog\n            |______ 00000128.jpg\n            |______ 00000181.bmp\n            |______ ...\n        |__ goddess\n            |______ 00000243.jpg\n            |______ 00000306.jpg\n            |______ ...\n        ...\n      images_out_folder: Path to write the images overlay with detected\n        landmarks. These images are useful when you need to debug accuracy\n        issues.\n      csvs_out_path: Path to write the CSV containing the detected landmark\n        coordinates and label of each image that can be used to train a pose\n        classification model.\n    \"\"\"\n    self._images_in_folder = images_in_folder\n    self._images_out_folder = images_out_folder\n    self._csvs_out_path = csvs_out_path\n    self._messages = []\n\n    # Create a temp dir to store the pose CSVs per class\n    self._csvs_out_folder_per_class = tempfile.mkdtemp()\n\n    # Get list of pose classes and print image statistics\n    self._pose_class_names = sorted(\n        [n for n in os.listdir(self._images_in_folder) if not n.startswith('.')]\n        )\n\n  def process(self, per_pose_class_limit=None, detection_threshold=0.1):\n    \"\"\"Preprocesses images in the given folder.\n    Args:\n      per_pose_class_limit: Number of images to load. As preprocessing usually\n        takes time, this parameter can be specified to make the reduce of the\n        dataset for testing.\n      detection_threshold: Only keep images with all landmark confidence score\n        above this threshold.\n    \"\"\"\n    # Loop through the classes and preprocess its images\n    for pose_class_name in self._pose_class_names:\n      print('Preprocessing', pose_class_name, file=sys.stderr)\n\n      # Paths for the pose class.\n      images_in_folder = os.path.join(self._images_in_folder, pose_class_name)\n      images_out_folder = os.path.join(self._images_out_folder, pose_class_name)\n      csv_out_path = os.path.join(self._csvs_out_folder_per_class,\n                                  pose_class_name + '.csv')\n      if not os.path.exists(images_out_folder):\n        os.makedirs(images_out_folder)\n\n      # Detect landmarks in each image and write it to a CSV file\n      with open(csv_out_path, 'w') as csv_out_file:\n        csv_out_writer = csv.writer(csv_out_file, \n                                    delimiter=',', \n                                    quoting=csv.QUOTE_MINIMAL)\n        # Get list of images\n        image_names = sorted(\n            [n for n in os.listdir(images_in_folder) if not n.startswith('.')])\n        if per_pose_class_limit is not None:\n          image_names = image_names[:per_pose_class_limit]\n\n        valid_image_count = 0\n\n        # Detect pose landmarks from each image\n        for image_name in tqdm.tqdm(image_names):\n          image_path = os.path.join(images_in_folder, image_name)\n\n          try:\n            image = tf.io.read_file(image_path)\n            image = tf.io.decode_jpeg(image)\n          except:\n            self._messages.append('Skipped ' + image_path + '. Invalid image.')\n            continue\n          else:\n            image = tf.io.read_file(image_path)\n            image = tf.io.decode_jpeg(image)\n            image_height, image_width, channel = image.shape\n\n          # Skip images that isn't RGB because Movenet requires RGB images\n          if channel != 3:\n            self._messages.append('Skipped ' + image_path +\n                                  '. Image isn\\'t in RGB format.')\n            continue\n          person = detect(image)\n\n          # Save landmarks if all landmarks were detected\n          min_landmark_score = min(\n              [keypoint.score for keypoint in person.keypoints])\n          should_keep_image = min_landmark_score >= detection_threshold\n          if not should_keep_image:\n            self._messages.append('Skipped ' + image_path +\n                                  '. No pose was confidentlly detected.')\n            continue\n\n          valid_image_count += 1\n\n          # Draw the prediction result on top of the image for debugging later\n          output_overlay = draw_prediction_on_image(\n              image.numpy().astype(np.uint8), person, \n              close_figure=True, keep_input_size=True)\n\n          # Write detection result into an image file\n          output_frame = cv2.cvtColor(output_overlay, cv2.COLOR_RGB2BGR)\n          cv2.imwrite(os.path.join(images_out_folder, image_name), output_frame)\n\n          # Get landmarks and scale it to the same size as the input image\n          pose_landmarks = np.array(\n              [[keypoint.coordinate.x, keypoint.coordinate.y, keypoint.score]\n                for keypoint in person.keypoints],\n              dtype=np.float32)\n\n          # Write the landmark coordinates to its per-class CSV file\n          coordinates = pose_landmarks.flatten().astype(np.str).tolist()\n          csv_out_writer.writerow([image_name] + coordinates)\n\n        if not valid_image_count:\n          raise RuntimeError(\n              'No valid images found for the \"{}\" class.'\n              .format(pose_class_name))\n\n    # Print the error message collected during preprocessing.\n    print('\\n'.join(self._messages))\n\n    # Combine all per-class CSVs into a single output file\n    all_landmarks_df = self._all_landmarks_as_dataframe()\n    all_landmarks_df.to_csv(self._csvs_out_path, index=False)\n\n  def class_names(self):\n    \"\"\"List of classes found in the training dataset.\"\"\"\n    return self._pose_class_names\n\n  def _all_landmarks_as_dataframe(self):\n    \"\"\"Merge all per-class CSVs into a single dataframe.\"\"\"\n    total_df = None\n    for class_index, class_name in enumerate(self._pose_class_names):\n      csv_out_path = os.path.join(self._csvs_out_folder_per_class,\n                                  class_name + '.csv')\n      per_class_df = pd.read_csv(csv_out_path, header=None)\n\n      # Add the labels\n      per_class_df['class_no'] = [class_index]*len(per_class_df)\n      per_class_df['class_name'] = [class_name]*len(per_class_df)\n\n      # Append the folder name to the filename column (first column)\n      per_class_df[per_class_df.columns[0]] = (os.path.join(class_name, '') \n        + per_class_df[per_class_df.columns[0]].astype(str))\n\n      if total_df is None:\n        # For the first class, assign its data to the total dataframe\n        total_df = per_class_df\n      else:\n        # Concatenate each class's data into the total dataframe\n        total_df = pd.concat([total_df, per_class_df], axis=0)\n\n    list_name = [[bodypart.name + '_x', bodypart.name + '_y', \n                  bodypart.name + '_score'] for bodypart in BodyPart] \n    header_name = []\n    for columns_name in list_name:\n      header_name += columns_name\n    header_name = ['file_name'] + header_name\n    header_map = {total_df.columns[i]: header_name[i] \n                  for i in range(len(header_name))}\n\n    total_df.rename(header_map, axis=1, inplace=True)\n\n    return total_df\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T11:56:34.059342Z","iopub.execute_input":"2023-07-28T11:56:34.059682Z","iopub.status.idle":"2023-07-28T11:56:34.082538Z","shell.execute_reply.started":"2023-07-28T11:56:34.059648Z","shell.execute_reply":"2023-07-28T11:56:34.081284Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"test_image_url = \"https://cdn.pixabay.com/photo/2017/03/03/17/30/yoga-2114512_960_720.jpg\"\n!wget -O /tmp/image.jpeg {test_image_url}\n\nif len(test_image_url):\n  image = tf.io.read_file('/tmp/image.jpeg')\n  image = tf.io.decode_jpeg(image)\n  person = detect(image)\n  _ = draw_prediction_on_image(image.numpy(), person, crop_region=None, \n                               close_figure=False, keep_input_size=True)\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T11:56:45.143469Z","iopub.execute_input":"2023-07-28T11:56:45.143911Z","iopub.status.idle":"2023-07-28T11:56:46.433607Z","shell.execute_reply.started":"2023-07-28T11:56:45.143858Z","shell.execute_reply":"2023-07-28T11:56:46.432757Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}