def _upload_labeling_task_annotations()

in backend/model_garden/management/commands/process_task_statuses.py [0:0]

• 43 lines of code
• 9 McCabe index (conditional complexity)

  def _upload_labeling_task_annotations(self, labeling_task: LabelingTask):
    dataset = labeling_task.media_assets.first().dataset
    annotation_frmt = self._get_annotations_format(dataset)

    try:
      annotations_content_zip = self._cvat_service.get_annotations(
        task_id=labeling_task.task_id,
        task_name=labeling_task.name,
        annotation_format=annotation_frmt,
      )
    except Exception as e:
      raise Exception(f"Failed to get task annotations: {e}")

    zip_fp = BytesIO(annotations_content_zip)
    zf = ZipFile(file=zip_fp)

    annotation_filenames = {
      os.path.split(zi.filename)[-1]: zf.open(zi) for zi in zf.filelist if
      zi.filename.startswith(self._get_cvat_zip_folderpath(annotation_frmt))
    }

    media_assets = labeling_task.media_assets.all()
    media_assets_filenames = {
      f"{os.path.splitext(media_asset.filename)[0]}" + self._get_label_file_extension(annotation_frmt) for media_asset
      in media_assets}
    logger.info(media_assets_filenames)
    logger.info(annotation_filenames)

    if len(annotation_filenames) == 0:
      raise NoAnnotationException(f"Missing all task annotations for task :{str(labeling_task.task_id)}")

    for media_asset in media_assets:
      try:
        asset_filename = os.path.splitext(media_asset.filename)[0]
        bucket_name = media_asset.dataset.bucket.name
        s3_client = S3Client(bucket_name=bucket_name)
        labeling_file_name = f"{asset_filename}" + self._get_label_file_extension(annotation_frmt)
        if labeling_file_name in annotation_filenames:
          file_object = annotation_filenames[f"{asset_filename}" + self._get_label_file_extension(annotation_frmt)]

          # TODO:remove deprecated remote_label_path property
          media_asset.labeling_asset_filepath = media_asset.remote_label_path
          s3_client.upload_file_obj(
            file_obj=file_object,
            bucket=bucket_name,
            key=media_asset.full_label_path,
          )
          media_asset.save()
        logger.info(f"Uploaded annotation '{media_asset.full_label_path}'")
      except Exception as e:
        raise Exception(f"Failed to upload task annotations: {e}")