Creates a progress file to notify NVCT that a new output is ready to upload :param percent_complete: value between 0 and 100 to demonstrate percent complete :param progress_file_path: progress file path :param progress_name: name of folder/artifact to upload to NGC :param metad
(percent_complete: int, progress_file_path: str,
progress_name: str, metadata: dict = None)
| 83 | |
| 84 | |
| 85 | def update_task_progress_file(percent_complete: int, progress_file_path: str, |
| 86 | progress_name: str, metadata: dict = None): |
| 87 | """ |
| 88 | Creates a progress file to notify NVCT that a new output is ready to upload |
| 89 | |
| 90 | :param percent_complete: value between 0 and 100 to demonstrate percent complete |
| 91 | :param progress_file_path: progress file path |
| 92 | :param progress_name: name of folder/artifact to upload to NGC |
| 93 | :param metadata: a dict of additional data to include |
| 94 | :return: |
| 95 | """ |
| 96 | # as requested in comment |
| 97 | if metadata is None: |
| 98 | metadata = dict() |
| 99 | |
| 100 | progress = Progress( |
| 101 | taskId=os.getenv("NVCT_TASK_ID", ""), |
| 102 | percentComplete=percent_complete, |
| 103 | name=progress_name, |
| 104 | lastUpdatedAt=datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), |
| 105 | metadata=metadata |
| 106 | ) |
| 107 | |
| 108 | temp_file = progress_file_path + ".tmp" |
| 109 | with open(temp_file, 'w') as file: |
| 110 | file.write(progress.output_to_json()) |
| 111 | |
| 112 | try: |
| 113 | os.rename(temp_file, progress_file_path) |
| 114 | except Exception as e: |
| 115 | print(f"Failed to write to progress file: {e}") |
| 116 | raise e |
| 117 | |
| 118 | |
| 119 | def create_file_with_specific_size(filename, size_in_bytes): |
no test coverage detected