def _post_downloading()

in src/nanotron/s3_checkpoints/s3_mover.py [0:0]

• 19 lines of code
• 10 McCabe index (conditional complexity)

    def _post_downloading(self) -> bool:
        self.get_current_stdout()
        s5cmd_results = [json.loads(i) for i in self.stdout.split("\n") if i]
        total_files = len([i for i in s5cmd_results if i["success"]])
        total_not_downloaded_files = len([i for i in s5cmd_results if not i["success"]])
        if total_not_downloaded_files == 0:
            all_upload = "all files"
            success = True
        else:
            all_upload = "not all files"
            success = False
        total_size = sum(i["object"]["size"] for i in s5cmd_results if "size" in i["object"])
        total_time = (datetime.now() - self.start_time).total_seconds()
        self._warning(
            f"[S3] Successfully downloaded {total_files} files for a total of {human_format(total_size)}B in {total_time}"
            f"sec ({all_upload}) from S3 at {self.s3_path} to {self.local_path}"
            f"(direct link: {self.s3_path_direct_link})"
        )
        return success