fix upload

openpose_gen.py

@@ -1,14 +1,13 @@
 import json
 import os
 import random
-import numpy as np
 from typing import List
-import math
 import cv2
+import numpy as np
+import urllib
 import skeleton_lib as skel
 import process_json_file as pjf
-import urllib
+from urllib import request
-from urllib import request, parse
 from requests_toolbelt.multipart.encoder import MultipartEncoder
 import sys
 import hashlib
@@ -16,76 +15,6 @@ sys.path.append('./')
 
 server_address = "localhost:8188"
 
-def is_normalized(keypoints: List[skel.Keypoint]) -> bool:
-    for keypoint in keypoints:
-        if not (0 <= keypoint.x <= 1 and 0 <= keypoint.y <= 1):
-            return False
-    return True
-
-def draw_bodypose(canvas: np.ndarray, keypoints: List[skel.Keypoint], limbSeq, colors, xinsr_stick_scaling: bool = False) -> np.ndarray:
-    """
-    Draw keypoints and limbs representing body pose on a given canvas.
-
-    Args:
-        canvas (np.ndarray): A 3D numpy array representing the canvas (image) on which to draw the body pose.
-        keypoints (List[Keypoint]): A list of Keypoint objects representing the body keypoints to be drawn.
-        xinsr_stick_scaling (bool): Whether or not scaling stick width for xinsr ControlNet
-
-    Returns:
-        np.ndarray: A 3D numpy array representing the modified canvas with the drawn body pose.
-
-    Note:
-        The function expects the x and y coordinates of the keypoints to be normalized between 0 and 1.
-    """
-    if not is_normalized(keypoints):
-        H, W = 1.0, 1.0
-    else:
-        H, W, _ = canvas.shape
-
-    CH, CW, _ = canvas.shape
-    stickwidth = 2
-
-    # Ref: https://huggingface.co/xinsir/controlnet-openpose-sdxl-1.0
-    max_side = max(CW, CH)
-    if xinsr_stick_scaling:
-        stick_scale = 1 if max_side < 500 else min(2 + (max_side // 1000), 7)
-    else :
-        stick_scale = 1
-    
-    if keypoints is None or len(keypoints) == 0:
-        return canvas
-
-    for (k1_index, k2_index), color in zip(limbSeq, colors):
-        keypoint1 = keypoints[k1_index]
-        keypoint2 = keypoints[k2_index]
-
-        if keypoint1 is None or keypoint2 is None or keypoint1.confidence == 0 or keypoint2.confidence == 0:
-            # if keypoint1 is None or keypoint1.confidence == 0:
-            #     print(f"keypoint failed: {k1_index}")
-            # if keypoint2 is None or keypoint2.confidence == 0:
-            #     print(f"keypoint failed: {k2_index}")
-            continue
-
-        Y = np.array([keypoint1.x, keypoint2.x]) * float(W)
-        X = np.array([keypoint1.y, keypoint2.y]) * float(H)
-        mX = np.mean(X)
-        mY = np.mean(Y)
-        length = ((X[0] - X[1]) ** 2 + (Y[0] - Y[1]) ** 2) ** 0.5
-        angle = math.degrees(math.atan2(X[0] - X[1], Y[0] - Y[1]))
-        polygon = cv2.ellipse2Poly((int(mY), int(mX)), (int(length / 2), stickwidth*stick_scale), int(angle), 0, 360, 1)
-        cv2.fillConvexPoly(canvas, polygon, [int(float(c) * 0.6) for c in color])
-
-    for keypoint, color in zip(keypoints, colors):
-        if keypoint is None or keypoint.confidence == 0:
-            continue
-
-        x, y = keypoint.x, keypoint.y
-        x = int(x * W)
-        y = int(y * H)
-        cv2.circle(canvas, (int(x), int(y)), 4, color, thickness=-1)
-
-    return canvas
-
 def coordinates_to_keypoints(coordinates: list) -> List[skel.Keypoint]:
     keypoints = [skel.Keypoint(coordinates[i], coordinates[i + 1]) for i in range(0, len(coordinates), 3)]
     return keypoints
@@ -96,9 +25,12 @@ def save_bodypose(width: int, height: int, coordinates: list, pid: str) -> None:
 
     canvas = np.zeros((height, width, 3), dtype=np.uint8)
     keypoints = coordinates_to_keypoints(coordinates)
-    canvas = draw_bodypose(canvas, keypoints, skel.coco_limbSeq, skel.coco_colors)
+    canvas = skel.draw_bodypose(canvas, keypoints, skel.coco_limbSeq, skel.coco_colors)
 
     # Save as body_pose_output0000.png, body_pose_output0001.png, ...
+    output_dir = 'output'
+    if not os.path.exists(output_dir):
+        os.makedirs(output_dir)
     image_path = 'output/body_pose_output%04d.png' % save_bodypose.counter
     cv2.imwrite(image_path, canvas)
     save_bodypose.counter += 1  # Increment the counter
@@ -112,9 +44,13 @@ def save_bodypose_mulit(width: int, height: int, coordinates_list: list, pid: st
     canvas = np.zeros((height, width, 3), dtype=np.uint8)
     for coordinates in coordinates_list:
         keypoints = coordinates_to_keypoints(coordinates)
-        canvas = draw_bodypose(canvas, keypoints, skel.coco_limbSeq, skel.coco_colors)
+        canvas = skel.draw_bodypose(canvas, keypoints, skel.coco_limbSeq, skel.coco_colors)
 
     # Save as body_pose_output0000.png, body_pose_output0001.png, ...
+    output_dir = 'output'
+    if not os.path.exists(output_dir):
+        os.makedirs(output_dir)
+        
     image_path = 'output/body_pose_output_multi%04d.png' % save_bodypose_mulit.counter
     cv2.imwrite(image_path, canvas)
     save_bodypose_mulit.counter += 1  # Increment the counter
@@ -127,7 +63,7 @@ def queue_prompt(prompt, server_address):
     req = request.Request("http://{}/prompt".format(server_address), data=data)
     request.urlopen(req)
 
-def upload_image(input_image, name, server_address, image_type="input", overwrite=False):
+def upload_image(input_image, name, server_address, image_type="input", overwrite=True):
     # Check if input_image is a valid file path
     if isinstance(input_image, str) and os.path.isfile(input_image):
         file = open(input_image, 'rb')
@@ -191,7 +127,7 @@ def main():
         for j, skeleton in enumerate(sliced):
             keypoints = skeleton.keypoints
             skeleton_sequences[j].get_frame(i).keypoints = keypoints
-            canvas = draw_bodypose(canvas, keypoints, skel.body_25_limbSeq, skel.body_25_colors)
+            canvas = skel.draw_bodypose(canvas, keypoints, skel.body_25_limbSeq, skel.body_25_colors)
 
         cv2.imwrite(image_path + '_' + str(i) + '.png', canvas)
 

process_json_file.py

@@ -1,7 +1,9 @@
 import os
 import json
 import sys
+import cv2
 import numpy as np
+import pandas as pd
 from typing import List
 import skeleton_lib as skel
 import concurrent.futures
@@ -45,7 +47,7 @@ def Skeleton_Seqences_save_to_array_json(skeleton_sequences: List[skel.Skeleton_
     with open(json_file, 'w') as file:
         json.dump(data, file, indent=4)
 
-def process_json_file(json_file, directory):
+def process_json_file(json_file, directory, output_directory):
     json_file = os.path.join(directory, json_file)
     # print(json_file)
 
@@ -65,20 +67,19 @@ def process_json_file(json_file, directory):
             keypoints = skel.fix_keypoints(keypoints, last_keypoints, next_keypoints)
             skeleton_sequences[j].get_frame(i).keypoints = keypoints
 
-    Skeleton_Seqences_save_to_array_json(skeleton_sequences, './fixed/' + os.path.basename(json_file))
+    Skeleton_Seqences_save_to_array_json(skeleton_sequences, output_directory + os.path.basename(json_file))
 
-def process_json_files_chunk(json_files_chunk, directory):
+def process_json_files_chunk(json_files_chunk, directory, output_directory):
     for json_file in json_files_chunk:
-        process_json_file(json_file, directory)
+        process_json_file(json_file, directory, output_directory)
 
-def process_json_files_multi_threaded(json_files, directory):
+def process_json_files_multi_threaded(json_files, directory, output_directory):
-    directory = './FencersKeyPoints'
     json_files = [f for f in os.listdir(directory) if f.endswith('.json')]
     if not json_files:
         print("No JSON files found in the directory.")
         return
 
-    json_files_chunks = np.array_split(json_files, 12)
+    json_files_chunks = np.array_split(json_files, 64)
 
     with concurrent.futures.ThreadPoolExecutor() as executor:
         futures = [executor.submit(process_json_files_chunk, chunk, directory) for chunk in json_files_chunks]
@@ -86,4 +87,46 @@ def process_json_files_multi_threaded(json_files, directory):
             try:
                 future.result()
             except Exception as e:
-                print(f"Error processing file chunk: {e}")
+                print(f"Error processing file chunk: {e}")
+
+def process_clip_descriptor(input_file_path, output_file_path):
+    ClipDescriptorKaggle = pd.read_csv(input_file_path)
+    
+    # Add a new column to store the YouTube video ID by extracting it from the URL
+    ClipDescriptorKaggle['video_id'] = ClipDescriptorKaggle['URL'].apply(lambda x: x.split('=')[1].split('&')[0])
+    
+    # Save the processed DataFrame to a new CSV file
+    with open(output_file_path, 'w') as file:
+        ClipDescriptorKaggle.to_csv(file, index=False)
+
+def get_frames_from_fixed_json(json_file):
+    frames = []
+    with open(json_file, 'r') as file:
+        data = json.load(file)
+        for frame in data:
+            skeletons = []
+            for i in range(2):  # Assuming there are always 2 skeletons
+                keypoints = []
+                for point in frame[i]:
+                    keypoint = skel.Keypoint(point[0], point[1], point[2])
+                    keypoints.append(keypoint)
+                skeletons.append(skel.Skeleton(keypoints))
+            frames.append(skeletons)
+
+    return frames
+
+def main():
+
+    descriptor = pd.read_csv('./ClipDescriptorKaggle_processed.csv')
+    
+    frames = get_frames_from_fixed_json('./fixed/0050_001_08_08_1.json')
+    # print(frames[0][0].keypoints[0])
+    
+    canvas = np.zeros((360, 640, 3), dtype=np.uint8)
+    canvas = skel.draw_bodypose(canvas, frames[0][0].keypoints, skel.body_25_limbSeq, skel.body_25_colors)
+    
+    #save the image
+    cv2.imwrite('test.png', canvas)
+    
+if __name__ == '__main__':
+    main()

skeleton_lib.py

@@ -1,4 +1,8 @@
 from typing import List
+import numpy as np
+import math
+import cv2
+from numpy import ndarray
 
 coco_limbSeq = [
     [1, 2], [1, 5], [2, 3], [3, 4], 
@@ -107,4 +111,76 @@ def fix_keypoints(keypoints, last_keypoints, next_keypoints):
     return keypoints
 
 def get_time_slice_for_Skeleton_Seqences(skeleton_seqences: List[Skeleton_Seqence], frame_index: int) -> List[Skeleton]:
-    return [skeleton_seq.get_frame(frame_index) for skeleton_seq in skeleton_seqences]
+    return [skeleton_seq.get_frame(frame_index) for skeleton_seq in skeleton_seqences]
+
+
+def is_normalized(keypoints: List[Keypoint]) -> bool:
+    for keypoint in keypoints:
+        if not (0 <= keypoint.x <= 1 and 0 <= keypoint.y <= 1):
+            return False
+    return True
+
+
+def draw_bodypose(canvas: ndarray, keypoints: List[Keypoint], limbSeq, colors, xinsr_stick_scaling: bool = False) -> np.ndarray:
+    """
+    Draw keypoints and limbs representing body pose on a given canvas.
+
+    Args:
+        canvas (np.ndarray): A 3D numpy array representing the canvas (image) on which to draw the body pose.
+        keypoints (List[Keypoint]): A list of Keypoint objects representing the body keypoints to be drawn.
+        xinsr_stick_scaling (bool): Whether or not scaling stick width for xinsr ControlNet
+
+    Returns:
+        np.ndarray: A 3D numpy array representing the modified canvas with the drawn body pose.
+
+    Note:
+        The function expects the x and y coordinates of the keypoints to be normalized between 0 and 1.
+    """
+    if not is_normalized(keypoints):
+        H, W = 1.0, 1.0
+    else:
+        H, W, _ = canvas.shape
+
+    CH, CW, _ = canvas.shape
+    stickwidth = 2
+
+    # Ref: https://huggingface.co/xinsir/controlnet-openpose-sdxl-1.0
+    max_side = max(CW, CH)
+    if xinsr_stick_scaling:
+        stick_scale = 1 if max_side < 500 else min(2 + (max_side // 1000), 7)
+    else :
+        stick_scale = 1
+    
+    if keypoints is None or len(keypoints) == 0:
+        return canvas
+
+    for (k1_index, k2_index), color in zip(limbSeq, colors):
+        keypoint1 = keypoints[k1_index]
+        keypoint2 = keypoints[k2_index]
+
+        if keypoint1 is None or keypoint2 is None or keypoint1.confidence == 0 or keypoint2.confidence == 0:
+            # if keypoint1 is None or keypoint1.confidence == 0:
+            #     print(f"keypoint failed: {k1_index}")
+            # if keypoint2 is None or keypoint2.confidence == 0:
+            #     print(f"keypoint failed: {k2_index}")
+            continue
+
+        Y = np.array([keypoint1.x, keypoint2.x]) * float(W)
+        X = np.array([keypoint1.y, keypoint2.y]) * float(H)
+        mX = np.mean(X)
+        mY = np.mean(Y)
+        length = ((X[0] - X[1]) ** 2 + (Y[0] - Y[1]) ** 2) ** 0.5
+        angle = math.degrees(math.atan2(X[0] - X[1], Y[0] - Y[1]))
+        polygon = cv2.ellipse2Poly((int(mY), int(mX)), (int(length / 2), stickwidth*stick_scale), int(angle), 0, 360, 1)
+        cv2.fillConvexPoly(canvas, polygon, [int(float(c) * 0.6) for c in color])
+
+    for keypoint, color in zip(keypoints, colors):
+        if keypoint is None or keypoint.confidence == 0:
+            continue
+
+        x, y = keypoint.x, keypoint.y
+        x = int(x * W)
+        y = int(y * H)
+        cv2.circle(canvas, (int(x), int(y)), 4, color, thickness=-1)
+
+    return canvas