\s*| \s*([A-Za-z0-9]+)'
+
+ # Search for the pattern in the HTML content
+ match = re.search(pattern, content, re.DOTALL)
+
+ if match:
+ serial_number = match.group(1)
+ return serial_number
+
+ return "?"
+
+def filterListOfStringsByRegex(string_list, regex_pattern):
+ import re
+ #Usage :
+ #input_list = ["apple", "banana", "cherry", "date", "elderberry"]
+ #pattern = r"^[a-c].*" # Matches strings starting with letters a, b, or c
+ #result = filterListOfStringsByRegex(input_list, pattern)
+
+ matched_strings = []
+
+ for string in string_list:
+ if re.match(regex_pattern, string):
+ matched_strings.append(string)
+
+ return matched_strings
+
+
+def convertListOfRegexToListOfLists(master_string_list,regex_list):
+ string_list_output = []
+ for regex_pattern in regex_list:
+ string_list_output.append(filterListOfStringsByRegex(master_string_list,regex_pattern))
+ return string_list_output
+
+
+"""
+Check if an entry is part of a given list
+"""
+def getEntryIndexInList(theList,theEntry):
+ i=0
+ for listItem in theList:
+ if(theEntry.lower()==listItem.lower()):
+ return i
+ i=i+1
+ return -1
+
+
+"""
+Check if an entry is in a sublist of our configuration
+"""
+def checkIfEntryIsInConfigurationKey(configuration,theKey,theEntry):
+ for listItem in configuration[theKey]:
+ if(theEntry==listItem): return 1
+ return 0
+
+"""
+Check if a joint is declared in the configuration hierarchy
+"""
+def getConfigurationJointIsDeclaredInHierarchy(configuration,theEntry):
+ #-------------------------------------------------------------------------------------------
+ if (theEntry=="everything"):
+ print(bcolors.WARNING,"EVERYTHING.. is declared always.. ",bcolors.ENDC)
+ return 1
+ #-------------------------------------------------------------------------------------------
+ try:
+ out = theEntry.split('_')
+ theEntry=out[0]
+ except:
+ print("getConfigurationJointIsDeclaredInHierarchy could not split ",theEntry)
+ #-------------------------------------------------------------------------------------------
+ if 'banned' in configuration:
+ for listItem in configuration['banned']:
+ if(theEntry.lower()==listItem['output'].lower()):
+ print(bcolors.WARNING," Joint ",theEntry," is declared in banlist! ",bcolors.ENDC)
+ return 1
+
+ if 'hierarchy' in configuration:
+ for listItem in configuration['hierarchy']:
+ #print("Check ",theEntry," vs ",listItem['joint'])
+ if(theEntry.lower()==listItem['joint'].lower()):
+ print("The Joint ",theEntry," is : declared in hierarchy")
+ return 1
+
+ print("Joint is not declared in hierarchy..")
+ #-------------------------------------------------------------------------------------------
+ return 0
+
+
+"""
+Retrieve the configuration joint priority of a joint is declared in the configuration hierarchy
+"""
+def getConfigurationJointPriority(configuration,theEntry):
+ #-------------------------------------------------------------------------------------------
+ if (theEntry=="everything"):
+ print("EVERYTHING.. has a high priority always.. ")
+ return 1
+ #-------------------------------------------------------------------------------------------
+ try:
+ out = theEntry.split('_')
+ theEntry=out[0]
+ except:
+ print("getConfigurationJointPriority could not split ",theEntry)
+ #-------------------------------------------------------------------------------------------
+ if 'outputMode' in configuration:
+ if (configuration['outputMode'] == "3d"):
+ print(bcolors.WARNING,"We treat all joints as terribly important in 3D point mode!",bcolors.ENDC)
+ return 1
+
+ if 'banned' in configuration:
+ for listItem in configuration['banned']:
+ if(theEntry==listItem['output']):
+ print(bcolors.WARNING," Joint ",theEntry," is in banlist! ",bcolors.ENDC)
+ return 0
+
+ if 'hierarchy' in configuration:
+ for listItem in configuration['hierarchy']:
+ if(theEntry==listItem['joint']):
+ print("The Importance of Joint ",theEntry," is : ",listItem['importance'])
+ return listItem['importance']
+
+ print("The Importance of Joint ",theEntry," is : 0 ")
+ #-------------------------------------------------------------------------------------------
+ return 0
+
+
+"""
+Get the parent network from our configuration joint hierarchy
+"""
+def getParentNetwork(configuration,theEntry):
+ #-------------------------------------------------------------------------------------------
+ if (theEntry=="everything"):
+ print("EVERYTHING.. has no parent.. ")
+ return "none"
+ #-------------------------------------------------------------------------------------------
+ try:
+ out = theEntry.split('_')
+ theJoint=out[0]
+ theChannel=out[1]
+ except:
+ print("getParentNetwork could not split ",theEntry)
+ theJoint=theEntry
+ theChannel=0
+ #-------------------------------------------------------------------------------------------
+ print("Checking the parent of Joint(",theJoint,")/Channel(",theChannel,")")
+ for listItem in configuration['hierarchy']:
+ if(theJoint==listItem['joint']):
+ if (listItem['inheritNetwork']=="none"):
+ return "none"
+ parentNetwork="%s_%s" % (listItem['inheritNetwork'],theChannel)
+ if (parentNetwork==theEntry):
+ return "none"
+ else:
+ return parentNetwork
+ #-------------------------------------------------------------------------------------------
+ return "none"
+
+
+if __name__ == '__main__':
+ print("Tools.py is a library!")
+ splitTextBasedOnGroupNumber(3,"tmp.tmp","tmpF.tmp")
+
diff --git a/src/python/mnet4/writeCSV.py b/src/python/mnet4/writeCSV.py
new file mode 100755
index 0000000..5494fd3
--- /dev/null
+++ b/src/python/mnet4/writeCSV.py
@@ -0,0 +1,33 @@
+#!/usr/bin/python3
+import numpy as np
+import csv
+import gc
+import time
+import array
+import sys
+
+
+def writeCSVFileHeader(filenameOutput,inputListLabels,inputStart,inputEnd):
+ inputNumber=0
+ fileCSV = open(filenameOutput,"w")
+ for entry in inputListLabels[inputStart:inputEnd]:
+ #if (inputNumber>=inputStart) and (inputNumber=inputStart) and (inputNumber 1:
+ for i in range(0, len(sys.argv)):
+ if sys.argv[i] == "--headless":
+ headless = True
+ if sys.argv[i] == "--live":
+ liveDemo = True
+ if sys.argv[i] == "--mt":
+ multiThreaded = True
+ if sys.argv[i] == "--calib":
+ calibrationFile = sys.argv[i+1]
+ if sys.argv[i] == "--focalLength":
+ fX = float(sys.argv[i+1])
+ fY = float(sys.argv[i+2])
+ alterFocalLength = True
+ if sys.argv[i] == "--frameskip":
+ frameSkip = int(sys.argv[i+1])
+ if sys.argv[i] == "--nnsubsample":
+ doNNEveryNFrames = int(sys.argv[i+1])
+ if sys.argv[i] == "--ik":
+ hcdLearningRate = float(sys.argv[i+1])
+ hcdEpochs = int(sys.argv[i+2])
+ hcdIterations = int(sys.argv[i+3])
+ if sys.argv[i] == "--smooth":
+ smoothingSampling = float(sys.argv[i+1])
+ smoothingCutoff = float(sys.argv[i+2])
+ if sys.argv[i] == "--noik":
+ doHCDPostProcessing = 0
+ if sys.argv[i] == "--aspectCorrection":
+ aspectCorrection = float(sys.argv[i+1])
+ if sys.argv[i] == "--noise":
+ addNoise = float(sys.argv[i+1])
+ if sys.argv[i] == "--size":
+ videoWidth = int(sys.argv[i+1])
+ videoHeight = int(sys.argv[i+2])
+ if sys.argv[i] == "--scale":
+ scale = float(sys.argv[i+1])
+ if sys.argv[i] == "--all":
+ doBody = True
+ doFace = True
+ doREye = True
+ doMouth = True
+ doHands = True
+ if sys.argv[i] == "--centercrop":
+ centerCrop = True
+ if sys.argv[i] == "--nobody":
+ doBody = False
+ if sys.argv[i] == "--face":
+ doFace = True
+ if sys.argv[i] == "--eyes" or sys.argv[i] == "--reye":
+ doREye = True
+ if sys.argv[i] == "--mouth":
+ doMouth = True
+ if sys.argv[i] == "--hands":
+ doHands = True
+ if sys.argv[i] == "--save":
+ saveVideo = True
+ if sys.argv[i] == "--engine":
+ engine = sys.argv[i+1]
+ print("Selecting MocapNET engine:", engine)
+ if sys.argv[i] == "--from":
+ videoFilePath = sys.argv[i+1]
+ if sys.argv[i] == "--profile":
+ doProfiling = True
+
+ if "/dev/video" in videoFilePath:
+ smoothingSampling = 20.0
+ smoothingCutoff = 8.0
+ print("Special low smoothing setup for live camera")
+
+ # ---------------------------------------------------------------------
+ # Initialize MocapNET
+ # ---------------------------------------------------------------------
+ print("\nInitializing MocapNET...")
+ mnet = easyMocapNETConstructor(
+ engine,
+ doProfiling = doProfiling,
+ multiThreaded = multiThreaded,
+ doHCDPostProcessing = doHCDPostProcessing,
+ hcdLearningRate = hcdLearningRate,
+ hcdEpochs = hcdEpochs,
+ hcdIterations = hcdIterations,
+ smoothingSampling = smoothingSampling,
+ smoothingCutoff = smoothingCutoff,
+ bvhScale = scale,
+ doBody = doBody,
+ doFace = doFace,
+ doREye = doREye,
+ doMouth = doMouth,
+ doHands = doHands,
+ addNoise = addNoise
+ )
+
+ if calibrationFile != "":
+ print("Enforcing Calibration file :", calibrationFile)
+ mnet.bvh.configureRendererFromFile(calibrationFile)
+
+ if alterFocalLength and fX is not None and fY is not None:
+ commands = dict()
+ commands["fX"] = fX
+ commands["fY"] = fY
+ mnet.bvh.configureRenderer(commands)
+
+ mnet.test()
+ # Record BVH to out.bvh (if not liveDemo, will be closed at end)
+ mnet.recordBVH(not liveDemo)
+
+ # ---------------------------------------------------------------------
+ # Initialize YMAPNet 2D Pose Estimator
+ # ---------------------------------------------------------------------
+ print("\nInitializing YMAPNet 2D pose estimator (PoseEstimator2D)...")
+ # These defaults match runYMAPNet.py
+ pose_threshold = 84
+ pose_keypoint_thresh = 60.0
+ pose_engine = "tensorflow" # YMAPNet backend
+ profiling_2d = False
+
+ estimator = PoseEstimator2D(
+ modelPath='2d_pose_estimation',
+ threshold=pose_threshold,
+ keypoint_threshold=pose_keypoint_thresh,
+ engine=pose_engine,
+ profiling=profiling_2d,
+ illustrate=False,
+ monitor=list(),
+ window_arrangement=list()
+ )
+
+ # ---------------------------------------------------------------------
+ # Capture device
+ # ---------------------------------------------------------------------
+ cap = getCaptureDeviceFromPath(videoFilePath, videoWidth, videoHeight)
+
+ # ---------------------------------------------------------------------
+ # Main loop
+ # ---------------------------------------------------------------------
+ print("\nStarting MocapNET using YMAPNet 2D Joint Estimator Input")
+ print("Press ESC or q to quit.\n")
+
+ frameNumber = 0
+ maxBrokenFrames = 100
+ brokenFrames = 0
+ totalProcessingTime = 0.0
+ totalProcessingSamples = 0
+
+ # For optional performance history plotting inside mnet
+ bvhAnglesForPlotting = list()
+
+ while cap.isOpened():
+ success, image = cap.read()
+
+ #Crop input to square before YMAPNet
+ if (centerCrop):
+ image = extract_centered_rectangle(image) #<- aspect ratio is not correct ..
+
+ # Frame skipping
+ if frameSkip > 0:
+ for _ in range(0, frameSkip):
+ frameNumber += 1
+ success, image = cap.read()
+
+ if not success or image is None:
+ brokenFrames += 1
+ eprint("Ignoring empty camera frame:", brokenFrames, "/", maxBrokenFrames)
+ if brokenFrames > maxBrokenFrames:
+ break
+ continue
+
+ # Reset broken frame counter on success
+ brokenFrames = 0
+
+ # Aspect ratio correction (optional)
+ if aspectCorrection != 1.0:
+ width = int(image.shape[1] * aspectCorrection)
+ height = int(image.shape[0])
+ image = cv2.resize(image, (width, height))
+
+ imageClean = image.copy()
+
+ # -----------------------------------------------------------------
+ # 2D Pose Estimation via YMAPNet (PoseEstimator2D)
+ # -----------------------------------------------------------------
+ start_2d = time.time()
+
+ # Run YMAPNet on current frame
+ estimator.process(image)
+
+ # If no skeletons detected, skip frame
+ if len(estimator.skeletons) == 0:
+ frameNumber += 1
+ continue
+
+ # This returns the exact format MocapNET expects:
+ # {'2dx_nose': ..., '2dy_nose': ..., '2dv_nose': ..., ... }
+ mocapNETInput = estimator.encodeSkeletonAsDict(0, denormalize=True, keypoint_names=mnet_names)
+ #print("mocapNETInput Before",mocapNETInput)
+ mocapNETInput = guessLandmarks(mocapNETInput)
+ mocapNETInput = fakeFeet(mocapNETInput)
+ #print("\n\nmocapNETInput ",mocapNETInput)
+
+ # -----------------------------------------------------------------
+ # IMPORTANT: match MocapNET's training aspect ratio, just like the
+ # MediaPipe-based pipeline used to do.
+ # -----------------------------------------------------------------
+ height, width = image.shape[:2]
+ currentAspectRatio = width / float(height)
+
+ for key in list(mocapNETInput.keys()):
+ if key.startswith("2dx_"):
+ part_name = key[4:] # strip "2dx_"
+ y_key = "2dy_" + part_name
+ if y_key in mocapNETInput:
+ x = mocapNETInput[key]
+ y = mocapNETInput[y_key]
+ # Apply the same normalization MediaPipe path used
+ x2, y2 = normalize2DPointWhileAlsoMatchingTrainingAspectRatio(
+ x, y, currentAspectRatio
+ )
+ mocapNETInput[key] = x2
+ mocapNETInput[y_key] = y2
+ # -----------------------------------------------------------------
+
+ # For visualization, draw YMAPNet annotations on a copy
+ annotated_image = image.copy()
+ # Do not open extra windows from PoseEstimator2D; MocapNET will handle showing
+ estimator.visualize(annotated_image, show=True, save=False)
+
+ end_2d = time.time()
+ mnet.hz_2DEst = secondsToHz(end_2d - start_2d)
+ mnet.history_hz_2DEst.append(mnet.hz_2DEst)
+ if len(mnet.history_hz_2DEst) > mnet.perfHistorySize:
+ mnet.history_hz_2DEst.pop(0)
+
+ # -----------------------------------------------------------------
+ # MocapNET 3D prediction
+ # -----------------------------------------------------------------
+ doNN = 1
+ if doNNEveryNFrames > 0:
+ doNN = (frameNumber % doNNEveryNFrames) == 0
+
+ start_total = time.time()
+ mocapNET3DOutput = mnet.predict3DJoints(mocapNETInput, runNN=doNN, runHCD=True)
+ mocapNETBVHOutput = mnet.outputBVH
+ bvhAnglesForPlotting.append(mocapNETBVHOutput)
+ if len(bvhAnglesForPlotting) > 100:
+ bvhAnglesForPlotting.pop(0)
+
+ # Update timing stats
+ totalProcessingTime += (time.time() - start_total)
+ totalProcessingSamples += 1
+
+ # -----------------------------------------------------------------
+ # Visualization
+ # -----------------------------------------------------------------
+ # visualizeMocapNETEnsemble returns (image, plotImage)
+ imageOut, plotImage = visualizeMocapNETEnsemble(
+ mnet,
+ annotated_image,
+ plotBVHChannels=False,
+ bvhAnglesForPlotting=bvhAnglesForPlotting
+ )
+
+ frameNumber += 1
+
+ # Print status line in terminal
+ mnet.printStatus()
+
+ # -----------------------------------------------------------------
+ # Optional saving of frames
+ # -----------------------------------------------------------------
+ if saveVideo:
+ cv2.imwrite('colorFrame_0_%05u.jpg' % (frameNumber), imageOut)
+
+ # -----------------------------------------------------------------
+ # Show window (unless headless)
+ # -----------------------------------------------------------------
+ if not headless:
+ cv2.imshow('MocapNET 4 using YMAPNet 2D Joints', imageOut)
+ key = cv2.waitKey(1) & 0xFF
+ if key == 27 or key == ord('q') or key == ord('Q'):
+ print("Terminating after receiving keyboard request")
+ break
+
+ # ---------------------------------------------------------------------
+ # Cleanup
+ # ---------------------------------------------------------------------
+ cap.release()
+ if not headless:
+ cv2.destroyAllWindows()
+
+ if totalProcessingSamples > 0:
+ avg_ms = 1000.0 * (totalProcessingTime / totalProcessingSamples)
+ avg_hz = secondsToHz(totalProcessingTime / totalProcessingSamples)
+ print("Average processing time over", totalProcessingSamples, "frames:")
+ print(" %0.1f ms (%0.1f Hz)" % (avg_ms, avg_hz))
+ else:
+ print("No processing time statistics (no valid frames processed).")
+
+ # Save video if requested
+ if saveVideo:
+ os.system(
+ "ffmpeg -framerate 30 -i colorFrame_0_%%05d.jpg "
+ "-s %ux%u -y -r 30 -pix_fmt yuv420p -threads 8 "
+ "livelastRun3DHiRes.mp4" % (videoWidth, videoHeight)
+ )
+
+ # Destroy MocapNET object so that out.bvh is finalized
+ del mnet
+
+
+# -------------------------------------------------------------------------
+# Entry point
+# -------------------------------------------------------------------------
+if __name__ == '__main__':
+ streamPosesFromCameraToMocapNET()
+
|