code: def assert_artifact_existence_and_validity(
      experiment_dir_cp: ExperimentDirCheckpoint,
      exists_for_driver_trials: bool,
      exists_for_worker_trials: bool,
      skip_validation: bool = False,
  ):
      for trial, trial_cp in experiment_dir_cp.trial_to_cps.items():
          artifact_data = trial_cp.artifact_data
          artifact_exists = artifact_data is not None
          # exists_for_xxx_trials == artifact_exists covers 2 cases:
          # 1. Artifact should not exist for driver trial + it actually doesn't exist
          # 2. Artifact should exist + it actually exists
          if trial.was_on_driver_node:
              assert exists_for_driver_trials == artifact_exists, (
                  "Trial {trial.trial_id} was ON THE DRIVER, where the artifact "
                  f"SHOULD {'' if exists_for_driver_trials else 'NOT'} exist, "
                  f"but found that it DOES {'' if artifact_exists else 'NOT'} exist.\n"
                  f"Directory: {experiment_dir_cp.dir}"
              )
          else:
              assert exists_for_worker_trials == artifact_exists, (
                  "Trial {trial.trial_id} was NOT ON THE DRIVER, where the artifact "
                  f"SHOULD {'' if exists_for_driver_trials else 'NOT'} exist, "
                  f"but found that it DOES {'' if artifact_exists else 'NOT'} exist.\n"
                  f"Directory: {experiment_dir_cp.dir}"
              )
  
          if not artifact_exists or skip_validation:
              continue
  
          # NOTE: This expects `artifact_data` to be comma-separated string
          # with the trial.config["id"] showing up as many times as the
          # latest checkpoint iteration.
          # Ex: checkpoint_5 is the latest checkpoint for trial with id=0
          # -> Expect artifact_data == "0,0,0,0,0,"
          artifact_data_list = artifact_data.split(",")[
              :-1
          ]  # Account for extra ,
          artifact_iter = len(artifact_data_list)
          checkpoint_iters = sorted(
              [
                  checkpoint_data["internal_iter"]
                  for _, checkpoint_data in trial_cp.checkpoints
              ],
              reverse=True,
          )
          # TODO(ml-team): Compare to latest checkpoint only after
          # checkpoint+artifact saving is done atomically.
          top_two = checkpoint_iters[:2]
          print(
              f"\nGot artifact_iter = {artifact_iter}, "
              f"and top 2 checkpoint iters were {top_two}"
          )
          trial_id = trial.config["id"]
          assert all(
              id == str(trial_id) for id in artifact_data_list
          ), f"The artifact data should contain only {trial_id}: {artifact_data_list}"
          assert artifact_iter >= min(top_two), (
              "The artifact data is not synced with respect to the latest checkpoint! "
              f"Expected the artifact to contain at least {min(top_two)} "
              f"iterations of data, but only got {artifact_iter}."
          )
  
imports: import argparse
  import csv
  import io
  import tarfile
  from dataclasses import dataclass
  import json
  import os
  import pickle
  import platform
  import re
  import shutil
  import signal
  import subprocess
  import tempfile
  import time
  from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
  import ray
  from ray.train._internal.storage import StorageContext
  from ray.tune.execution.experiment_state import _find_newest_experiment_checkpoint
  from ray.tune.result import _get_defaults_results_dir
  from ray.tune.utils.serialization import TuneFunctionDecoder
label_desc: Use image node_id 27 for calling a global function with example usage: assert_artifact_existence_and_validity(experiment_dir_cp, exists_for_driver_trials, exists_for_worker_trials, skip_validation) without return types