in deepracer_systems_pkg/deepracer_systems_pkg/model_loader_module/model_loader_node.py [0:0]
def verify_model_ready(self, model_name):
"""Helper function to wait for model loading to complete if its not already,
else verify if the model folder has a checksum file created.
Args:
model_name (str): Model folder name in /opt/aws/deepracer/artifacts folder.
Returns:
bool: True if the model is properly loaded/has checksum.txt in the model folder else False.
"""
self.get_logger().info(f"Verifying readiness of model \"{model_name}\"...")
try:
with utility.AutoLock(self.progress_guard):
state = self.models_in_progress[model_name]
try:
# Wait for the model to become ready.
self.get_logger().info(f"Making sure \"{model_name}\" thread is complete...")
state.wait_complete()
# Return install status.
ready = state.installed.isSet()
if ready:
self.get_logger().info(f"Model \"{model_name}\" is successfully installed.")
else:
self.get_logger().info(f"Model \"{model_name}\" installation failed.")
except Exception as ex:
self.get_logger().error(f"Failed to wait for installation of \"{model_name}\": {ex}")
ready = False
except Exception as ex:
self.get_logger().info(f"Model \"{model_name}\" not being currently installed,"
" checking existing installation...")
model_install_directory = os.path.join(model_loader_config.MODEL_INSTALL_ROOT_DIRECTORY, model_name)
if not os.path.isdir(model_install_directory):
self.get_logger().info(f"Model \"{model_name}\" name is not recognized.")
ready = False
else:
checksum_path = os.path.join(model_install_directory, model_loader_config.MODEL_CHECKSUM_FILE)
checksum = file_system_utils.read_line(checksum_path).strip()
ready = (checksum != "")
if ready:
self.get_logger().info(f"Model \"{model_name}\" is installed.")
else:
self.get_logger().info(f"Model \"{model_name}\" installation appears incomplete.")
return ready