Merge pull request #169 from stanfordnmbl/missing_cameras

[WIP] missing cameras

app.py

@@ -121,10 +121,21 @@
                          headers = {"Authorization": "Token {}".format(API_TOKEN)})
         continue
 
-    if any([v["video"] is None for v in trial["videos"]]):
-        r = requests.patch(trial_url, data={"status": "error"},
-                     headers = {"Authorization": "Token {}".format(API_TOKEN)})
-        continue
+    # This is a hack to have the trials with status "reprocess" to be reprocessed
+    # with camerasToUse_c = ['all_available'] instead of ['all']. In practice, this
+    # allows reprocessing on server trials that failed because video(s) were not available.
+    # This is a temporary solution until we have a better way to handle this. By default,
+    # trials with missing videos are error-ed out directly so that we do not spend time
+    # on processing them.
+    status = trial["status"]
+    if status == "reprocess":
+        camerasToUse_c = ['all_available']
+    else:
+        camerasToUse_c = ['all']
+        if any([v["video"] is None for v in trial["videos"]]):
+            r = requests.patch(trial_url, data={"status": "error"},
+                        headers = {"Authorization": "Token {}".format(API_TOKEN)})
+            continue
 
     trial_type = "dynamic"
     if trial["name"] == "calibration":
@@ -137,7 +148,7 @@
 
     try:
         # trigger reset of timer for last processed trial                            
-        processTrial(trial["session"], trial["id"], trial_type=trial_type, isDocker=isDocker)   
+        processTrial(trial["session"], trial["id"], trial_type=trial_type, isDocker=isDocker, camerasToUse=camerasToUse_c)   
         # note a result needs to be posted for the API to know we finished, but we are posting them 
         # automatically thru procesTrial now
         r = requests.patch(trial_url, data={"status": "done"},

main.py

@@ -316,13 +316,59 @@ def main(sessionName, trialName, trial_id, camerasToUse=['all'],
         else:
             raise Exception('checkerBoard placement value in\
              sessionMetadata.yaml is not currently supported')
+
+        # Detect all available cameras (ie, cameras with existing videos).
+        cameras_available = []
+        for camName in cameraDirectories:
+            camDir = cameraDirectories[camName]
+            pathVideoWithoutExtension = os.path.join(camDir, 'InputMedia', trialName, trial_id)
+            if len(glob.glob(pathVideoWithoutExtension + '*')) == 0:
+                print(f"Camera {camName} does not have a video for trial {trial_id}")
+            else:
+                if os.path.exists(os.path.join(pathVideoWithoutExtension + getVideoExtension(pathVideoWithoutExtension))):
+                    cameras_available.append(camName)
+                else:
+                    print(f"Camera {camName} does not have a video for trial {trial_id}")
+
+        if camerasToUse[0] == 'all':
+            cameras_all = list(cameraDirectories.keys())
+            if not all([cam in cameras_available for cam in cameras_all]):
+                exception = 'Not all cameras have uploaded videos; one or more cameras might have turned off or lost connection'
+                raise Exception(exception, exception)
+            else:
+                camerasToUse_c = camerasToUse
+        elif camerasToUse[0] == 'all_available':
+            camerasToUse_c = cameras_available
+            print(f"Using available cameras: {camerasToUse_c}")
+        else:
+            if not all([cam in cameras_available for cam in camerasToUse]):
+                raise Exception('Not all specified cameras in camerasToUse have videos; verify the camera names or consider setting camerasToUse to ["all_available"]')
+            else:
+                camerasToUse_c = camerasToUse
+                print(f"Using cameras: {camerasToUse_c}")
+        settings['camerasToUse'] = camerasToUse_c
+        if camerasToUse_c[0] != 'all' and len(camerasToUse_c) < 2:
+            exception = 'At least two videos are required for 3D reconstruction, video upload likely failed for one or more cameras.'
+            raise Exception(exception, exception)
+
+        # For neutral, we do not allow reprocessing with not all cameras.
+        # The reason is that it affects extrinsics selection, and then you can only process
+        # dynamic trials with the same camera selection (ie, potentially not all cameras). 
+        # This might be addressable, but I (Antoine) do not see an immediate need + this
+        # would be a significant change in the code base. In practice, a data collection
+        # will not go through neutral if not all cameras are available.
+        if scaleModel:
+            if camerasToUse_c[0] != 'all' and len(camerasToUse_c) < len(cameraDirectories):
+                exception = 'All cameras are required for calibration and neutral pose.'
+                raise Exception(exception, exception)
+
         # Run pose detection algorithm.
         try:        
             videoExtension = runPoseDetector(
                     cameraDirectories, trialRelativePath, poseDetectorDirectory,
                     trialName, CamParamDict=CamParamDict, 
                     resolutionPoseDetection=resolutionPoseDetection, 
-                    generateVideo=generateVideo, cams2Use=camerasToUse,
+                    generateVideo=generateVideo, cams2Use=camerasToUse_c,
                     poseDetector=poseDetector, bbox_thr=bbox_thr)
             trialRelativePath += videoExtension
         except Exception as e:
@@ -335,14 +381,14 @@ def main(sessionName, trialName, trial_id, camerasToUse=['all'],
                 raise Exception(exception, traceback.format_exc())
 
     if runSynchronization:
-        # Synchronize videos. 
+        # Synchronize videos.
         try:
             keypoints2D, confidence, keypointNames, frameRate, nansInOut, startEndFrames, cameras2Use = (
                 synchronizeVideos( 
                     cameraDirectories, trialRelativePath, poseDetectorDirectory,
                     undistortPoints=True, CamParamDict=CamParamDict,
                     filtFreqs=filtFreqs, confidenceThreshold=0.4,
-                    imageBasedTracker=False, cams2Use=camerasToUse, 
+                    imageBasedTracker=False, cams2Use=camerasToUse_c, 
                     poseDetector=poseDetector, trialName=trialName,
                     resolutionPoseDetection=resolutionPoseDetection))
         except Exception as e:
@@ -357,6 +403,14 @@ def main(sessionName, trialName, trial_id, camerasToUse=['all'],
                     potential causes for a failed trial."""
                 raise Exception(exception, traceback.format_exc())
 
+    # Note: this should not be necessary, because we prevent reprocessing the neutral trial
+    # with not all cameras, but keeping it in there in case we would want to.
+    if calibrationOptions is not None:
+        allCams = list(calibrationOptions.keys())
+        for cam_t in allCams:
+            if not cam_t in cameras2Use:
+                calibrationOptions.pop(cam_t)
+
     if scaleModel and calibrationOptions is not None and alternateExtrinsics is None:
         # Automatically select the camera calibration to use
         CamParamDict = autoSelectExtrinsicSolution(sessionDir,keypoints2D,confidence,calibrationOptions)
@@ -549,7 +603,8 @@ def main(sessionName, trialName, trial_id, camerasToUse=['all'],
                                vertical_offset=vertical_offset)  
 
     # %% Rewrite settings, adding offset  
-    if not extrinsicsTrial and offset:
-        settings['verticalOffset'] = vertical_offset_settings 
+    if not extrinsicsTrial:
+        if offset:
+            settings['verticalOffset'] = vertical_offset_settings 
         with open(pathSettings, 'w') as file:
             yaml.dump(settings, file)

utils.py

@@ -770,9 +770,11 @@ def postMotionData(trial_id,session_path,trial_name=None,isNeutral=False,
     camDirs = glob.glob(os.path.join(session_path,'Videos','Cam*'))
     for camDir in camDirs:
         outputPklFolder = os.path.join(camDir,pklDir)
-        pklPath = glob.glob(os.path.join(outputPklFolder,'*_pp.pkl'))[0]
-        _,camName = os.path.split(camDir)
-        postFileToTrial(pklPath,trial_id,tag='pose_pickle',device_id=camName)
+        pickle_files = glob.glob(os.path.join(outputPklFolder,'*_pp.pkl'))
+        if pickle_files:
+            pklPath = pickle_files[0]
+            _,camName = os.path.split(camDir)
+            postFileToTrial(pklPath,trial_id,tag='pose_pickle',device_id=camName)
 
     # post marker data
     deleteResult(trial_id, tag='marker_data')

utilsServer.py

@@ -40,7 +40,8 @@ def processTrial(session_id, trial_id, trial_type = 'dynamic',
                  deleteLocalFolder = True,
                  hasWritePermissions = True,
                  use_existing_pose_pickle = False,
-                 batchProcess = False):
+                 batchProcess = False,
+                 camerasToUse=['all']):
 
     # Get session directory
     session_name = session_id 
@@ -61,7 +62,8 @@ def processTrial(session_id, trial_id, trial_type = 'dynamic',
         # run calibration
         try:
             main(session_name, trial_name, trial_id, isDocker=isDocker, extrinsicsTrial=True,
-                 imageUpsampleFactor=imageUpsampleFactor,genericFolderNames = True)
+                 imageUpsampleFactor=imageUpsampleFactor,genericFolderNames = True,
+                 camerasToUse=camerasToUse)
         except Exception as e:
             error_msg = {}
             error_msg['error_msg'] = e.args[0]
@@ -122,7 +124,8 @@ def processTrial(session_id, trial_id, trial_type = 'dynamic',
                  resolutionPoseDetection = resolutionPoseDetection,
                  genericFolderNames = True,
                  bbox_thr = bbox_thr,
-                 calibrationOptions = calibrationOptions)
+                 calibrationOptions = calibrationOptions,
+                 camerasToUse=camerasToUse)
         except Exception as e:       
             # Try to post pose pickles so can be used offline. This function will 
             # error at kinematics most likely, but if pose estimation completed,
@@ -211,7 +214,8 @@ def processTrial(session_id, trial_id, trial_type = 'dynamic',
                  imageUpsampleFactor=imageUpsampleFactor,
                  resolutionPoseDetection = resolutionPoseDetection,
                  genericFolderNames = True,
-                 bbox_thr = bbox_thr)
+                 bbox_thr = bbox_thr,
+                 camerasToUse=camerasToUse)
         except Exception as e:
             # Try to post pose pickles so can be used offline. This function will 
             # error at kinematics most likely, but if pose estimation completed,
@@ -331,7 +335,8 @@ def newSessionSameSetup(session_id_old,session_id_new,extrinsicTrialName='calibr
 
 def batchReprocess(session_ids,calib_id,static_id,dynamic_trialNames,poseDetector='OpenPose', 
                    resolutionPoseDetection='1x736',deleteLocalFolder=True,
-                   isServer=False, use_existing_pose_pickle=True):
+                   isServer=False, use_existing_pose_pickle=True,
+                   camerasToUse=['all']):
 
     # extract trial ids from trial names
     if dynamic_trialNames is not None and len(dynamic_trialNames)>0:
@@ -366,7 +371,8 @@ def batchReprocess(session_ids,calib_id,static_id,dynamic_trialNames,poseDetecto
                               poseDetector = poseDetector,
                               deleteLocalFolder = deleteLocalFolder,
                               isDocker=isServer,
-                              hasWritePermissions = hasWritePermissions)
+                              hasWritePermissions = hasWritePermissions,
+                              camerasToUse=camerasToUse)
                 statusData = {'status':'done'}
                 _ = requests.patch(API_URL + "trials/{}/".format(calib_id_toProcess), data=statusData,
                          headers = {"Authorization": "Token {}".format(API_TOKEN)})
@@ -392,7 +398,8 @@ def batchReprocess(session_ids,calib_id,static_id,dynamic_trialNames,poseDetecto
                               isDocker=isServer,
                               hasWritePermissions = hasWritePermissions,
                               use_existing_pose_pickle = use_existing_pose_pickle,
-                              batchProcess = True)
+                              batchProcess = True,
+                              camerasToUse=camerasToUse)
                 statusData = {'status':'done'}
                 _ = requests.patch(API_URL + "trials/{}/".format(static_id_toProcess), data=statusData,
                          headers = {"Authorization": "Token {}".format(API_TOKEN)})
@@ -423,7 +430,8 @@ def batchReprocess(session_ids,calib_id,static_id,dynamic_trialNames,poseDetecto
                           isDocker=isServer,
                           hasWritePermissions = hasWritePermissions,
                           use_existing_pose_pickle = use_existing_pose_pickle,
-                          batchProcess = True)
+                          batchProcess = True,
+                          camerasToUse=camerasToUse)
 
                 statusData = {'status':'done'}
                 _ = requests.patch(API_URL + "trials/{}/".format(dID), data=statusData,