{"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\nfrom matplotlib import pyplot as plt\nfrom matplotlib.collections import LineCollection\n\nimport tensorflow as tf\nimport tensorflow_hub as hub\nfrom tensorflow import keras\n\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.metrics import accuracy_score, classification_report, confusion_matrix\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T14:35:43.275307Z","iopub.execute_input":"2023-07-28T14:35:43.275657Z","iopub.status.idle":"2023-07-28T14:35:53.553611Z","shell.execute_reply.started":"2023-07-28T14:35:43.275628Z","shell.execute_reply":"2023-07-28T14:35:53.552693Z"},"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":{"execution":{"iopub.status.busy":"2023-07-28T14:35:53.554927Z","iopub.execute_input":"2023-07-28T14:35:53.555778Z","iopub.status.idle":"2023-07-28T14:36:02.951839Z","shell.execute_reply.started":"2023-07-28T14:35:53.555739Z","shell.execute_reply":"2023-07-28T14:36:02.950486Z"},"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-28T14:36:02.955550Z","iopub.execute_input":"2023-07-28T14:36:02.956034Z","iopub.status.idle":"2023-07-28T14:36:02.964348Z","shell.execute_reply.started":"2023-07-28T14:36:02.955986Z","shell.execute_reply":"2023-07-28T14:36:02.963187Z"},"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-28T14:36:02.969334Z","iopub.execute_input":"2023-07-28T14:36:02.969723Z","iopub.status.idle":"2023-07-28T14:36:03.002178Z","shell.execute_reply.started":"2023-07-28T14:36:02.969692Z","shell.execute_reply":"2023-07-28T14:36:03.001007Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"is_skip_step_1 = False\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T14:36:03.003703Z","iopub.execute_input":"2023-07-28T14:36:03.004154Z","iopub.status.idle":"2023-07-28T14:36:03.023988Z","shell.execute_reply.started":"2023-07-28T14:36:03.004109Z","shell.execute_reply":"2023-07-28T14:36:03.022573Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"use_custom_dataset = True\n\ndataset_is_split = True\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T14:36:03.025237Z","iopub.execute_input":"2023-07-28T14:36:03.025600Z","iopub.status.idle":"2023-07-28T14:36:03.036176Z","shell.execute_reply.started":"2023-07-28T14:36:03.025568Z","shell.execute_reply":"2023-07-28T14:36:03.035187Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"dataset_in = '/kaggle/input/ai2022-competition/Test/'\n\nif dataset_is_split:\n    IMAGES_ROOT = dataset_in\nelse:\n    dataset_out = 'split_' + dataset_in\n    split_into_train_test(dataset_in, dataset_out, test_split=0.2)\n    IMAGES_ROOT = dataset_out","metadata":{"execution":{"iopub.status.busy":"2023-07-28T14:41:38.953061Z","iopub.execute_input":"2023-07-28T14:41:38.953554Z","iopub.status.idle":"2023-07-28T14:41:38.960529Z","shell.execute_reply.started":"2023-07-28T14:41:38.953513Z","shell.execute_reply":"2023-07-28T14:41:38.959192Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"if not is_skip_step_1:\n  images_in_test_folder = os.path.join(IMAGES_ROOT)\n  images_out_test_folder = 'poses_images_out_test'\n  csvs_out_test_path = 'test_data.csv'\n\n  preprocessor = MoveNetPreprocessor(\n      images_in_folder=images_in_test_folder,\n      images_out_folder=images_out_test_folder,\n      csvs_out_path=csvs_out_test_path,\n  )\n\n  preprocessor.process(per_pose_class_limit=None)\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T14:41:39.483135Z","iopub.execute_input":"2023-07-28T14:41:39.484107Z","iopub.status.idle":"2023-07-28T14:54:22.514573Z","shell.execute_reply.started":"2023-07-28T14:41:39.484068Z","shell.execute_reply":"2023-07-28T14:54:22.513229Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import zipfile\n\n# Define the path of the folder you want to download\nfolder_path = '/kaggle/working/poses_images_out_test'\n\n# Define the path of the .zip file you want to create\nzip_file_path = '/kaggle/working/output_test.zip'\n\n# Create a .zip file containing the contents of the folder\nwith zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED) as zipf:\n    for root, _, files in os.walk(folder_path):\n        for file in files:\n            file_path = os.path.join(root, file)\n            arcname = os.path.relpath(file_path, folder_path)\n            zipf.write(file_path, arcname)\n\nprint(f\"Folder '{folder_path}' has been compressed and saved as '{zip_file_path}'.\")\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T14:54:22.517322Z","iopub.execute_input":"2023-07-28T14:54:22.517936Z","iopub.status.idle":"2023-07-28T14:54:55.815146Z","shell.execute_reply.started":"2023-07-28T14:54:22.517888Z","shell.execute_reply":"2023-07-28T14:54:55.813767Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import IPython\nIPython.display.FileLink('output_test.zip')\n","metadata":{"execution":{"iopub.status.busy":"2023-07-28T14:54:55.816447Z","iopub.execute_input":"2023-07-28T14:54:55.816883Z","iopub.status.idle":"2023-07-28T14:54:55.826279Z","shell.execute_reply.started":"2023-07-28T14:54:55.816853Z","shell.execute_reply":"2023-07-28T14:54:55.825171Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}