Small fixes from local branch

mephisto/abstractions/architects/ec2/ec2_helpers.py

@@ -904,7 +904,10 @@ def try_server_push(subprocess_args: List[str], retries=5, sleep_time=10.0):
         except subprocess.CalledProcessError:
             retries -= 1
             sleep_time *= 1.5
-            logger.info(f"Timed out trying to push to server. Retries remaining: {retries}")
+            logger.info(
+                f"Timed out trying to push to server. CMD: {subprocess_args}. "
+                f"Retries remaining: {retries}"
+            )
             time.sleep(sleep_time)
     raise Exception("Could not successfully push to the ec2 instance. See log for errors.")
 

mephisto/abstractions/blueprints/remote_procedure/remote_procedure_blueprint.py

@@ -101,6 +101,15 @@ class RemoteProcedureBlueprintArgs(
             "required": True,
         },
     )
+    extra_source_dir: str = field(
+        default=MISSING,
+        metadata={
+            "help": (
+                "Optional path to sources that the HTML may "
+                "refer to (such as images/video/css/scripts)"
+            ),
+        },
+    )
     link_task_source: bool = field(
         default=False,
         metadata={

mephisto/data_model/agent.py

@@ -254,7 +254,7 @@ def _raise_if_disconnected():
             if not self.did_submit.is_set():
                 # If released without submit, raise timeout
                 raise AgentTimeoutError(timeout, self.db_id)
-            # Check disconnect possiblities again
+            # Check disconnect possibilities again
             _raise_if_disconnected()
         return self.did_submit.is_set()
 
@@ -456,7 +456,7 @@ def get_data_dir(self) -> str:
         this agent into
         """
         assignment_dir = self.get_assignment().get_data_dir()
-        return os.path.join(assignment_dir, self.db_id)
+        return os.path.join(assignment_dir, str(self.db_id))
 
     def update_status(self, new_status: str) -> None:
         """Update the database status of this agent, and

mephisto/operations/operator.py

@@ -386,9 +386,10 @@ def cleanup_runs():
             if not shutdown_thread.is_alive():
                 # Only join if the shutdown fully completed
                 shutdown_thread.join()
-        if self._event_loop.is_running():
-            self._event_loop.stop()
-        self._event_loop.run_until_complete(self.shutdown_async())
+        if not self._event_loop.is_running():
+            self._event_loop.run_until_complete(self.shutdown_async())
+        else:
+            asyncio.ensure_future(self.shutdown_async(), loop=self._event_loop)
 
     async def shutdown_async(self):
         """Shut down the asyncio parts of the Operator"""
@@ -517,7 +518,8 @@ async def _stop_loop_when_no_running_tasks(self, log_rate: Optional[int] = None)
                     last_log = time.time()
                     self.print_run_details()
             await asyncio.sleep(RUN_STATUS_POLL_TIME)
-        self._event_loop.stop()
+        if not self.is_shutdown:
+            self.shutdown()
 
     def _run_loop_until(self, condition_met: Callable[[], bool], timeout) -> bool:
         """

mephisto/tools/examine_utils.py

@@ -121,7 +121,10 @@ def get_worker_stats(units: List["Unit"]) -> Dict[str, Dict[str, List["Unit"]]]:
                 "soft_rejected": [],
                 "rejected": [],
             }
-        previous_work_by_worker[w_id][unit.get_status()].append(unit)
+        status = unit.get_status()
+        if status not in previous_work_by_worker[w_id]:
+            continue
+        previous_work_by_worker[w_id][status].append(unit)
     return previous_work_by_worker
 
 
@@ -250,13 +253,13 @@ def run_examine_by_worker(
                     should_special_qualify = input(
                         "Do you want to approve qualify this worker? (y)es/(n)o: "
                     )
-                    if should_special_qualify.lower() in ["y", "yes"]:
+                    if should_special_qualify.lower().startswith("y"):
                         worker.grant_qualification(approve_qualification, 1)
             elif decision.lower() == "p":
                 agent.soft_reject_work()
                 if apply_all_decision is None and block_qualification is not None:
                     should_soft_block = input("Do you want to soft block this worker? (y)es/(n)o: ")
-                    if should_soft_block.lower() in ["y", "yes"]:
+                    if should_soft_block.lower().startswith("y"):
                         worker.grant_qualification(block_qualification, 1)
             elif decision.lower() == "v":
                 # Same as "a", except we can specify exact qualification value being assigned
@@ -271,7 +274,7 @@ def run_examine_by_worker(
                 if apply_all_decision is None:
                     reason = input("Why are you rejecting this work? ")
                     should_block = input("Do you want to hard block this worker? (y)es/(n)o: ")
-                    if should_block.lower() in ["y", "yes"]:
+                    if should_block.lower().startswith("y"):
                         block_reason = input("Why permanently block this worker? ")
                         worker.block_worker(block_reason, unit=unit)
                 agent.reject_work(reason)