code: def assert_true_cdf_equal_by_dkwm(
      samples,
      cdf,
      left_continuous_cdf=None,
      false_fail_rate=1e-6,
      name=None,
  ):
      """Asserts the full CDF of the given distribution is as expected.
  
      More precisely, fails if there is enough evidence (using the
      [Dvoretzky-Kiefer-Wolfowitz-Massart inequality]
      (https://en.wikipedia.org/wiki/CDF-based_nonparametric_confidence_interval))
      that the true CDF of some distribution from which the given samples are
      drawn is _not_ the given expected CDF with statistical significance
      `false_fail_rate` or stronger, otherwise passes.  If you also want to
      check that you are gathering enough evidence that a pass is not
      spurious, see `min_num_samples_for_dkwm_cdf_test` and
      `min_discrepancy_of_true_cdfs_detectable_by_dkwm`.
  
      If the distribution in question has atoms (e.g., is discrete), computing this
      test requires CDF values for both sides of the discontinuity.  In this case,
      the `cdf` argument is assumed to compute the CDF inclusive of the atom, i.e.,
      cdf(x) = Pr(X <= x).  The user must also supply the `left_continuous_cdf`,
      which must compute the cdf exclusive of the atom, i.e., left_continuous_cdf(x)
      = Pr(X < x).  Invariant: cdf(x) - left_continuous_cdf(x) = pmf(x).
  
      For example, the two required cdfs of the degenerate distribution that places
      all the mass at 0 can be given as
      ```
      cdf=lambda x: tf.where(x < 0, 0., 1.)
      left_continuous_cdf=lambda x: tf.where(x <= 0, 0., 1.)
      ```
  
      Note that `false_fail_rate` is a total false failure rate for all
      the assertions in the batch.  As such, if the batch is nontrivial,
      the assertion will insist on stronger evidence to fail any one member.
  
      Args:
        samples: Tensor of shape [n] + B.  Samples from some (batch of) scalar-event
          distribution(s) of interest, giving a (batch of) empirical CDF(s).
          Assumed IID across the 0 dimension.
        cdf: Analytic cdf inclusive of any atoms, as a function that can compute CDF
          values in batch.  Must accept a Tensor of shape [n] + B and the same dtype
          as `samples` and return a Tensor of shape [n] + B of CDF values.  For each
          sample x, `cdf(x) = Pr(X <= x)`.
        left_continuous_cdf: Analytic left-continuous cdf, as a function that can
          compute CDF values in batch.  Must accept a Tensor of shape [n] + B and
          the same dtype as `samples` and return a Tensor of shape [n] + B of CDF
          values.  For each sample x, `left_continuous_cdf(x) = Pr(X < x)`.  If the
          distribution under test has no atoms (i.e., the CDF is continuous), this
          is redundant and may be omitted.  Conversely, if this argument is omitted,
          the test assumes the distribution is atom-free.
        false_fail_rate: *Scalar* floating-point `Tensor` admissible total
          rate of mistakes.
        name: A name for this operation (optional).
  
      Returns:
        check: Op that raises `InvalidArgumentError` if any expected CDF is
          outside the corresponding confidence envelope.
      """
      with tf.name_scope(name or "assert_true_cdf_equal_by_dkwm"):
          dtype = dtype_util.common_dtype(
              [samples, false_fail_rate], tf.float32
          )
          samples = tf.convert_to_tensor(
              value=samples, name="samples", dtype=dtype
          )
          false_fail_rate = tf.convert_to_tensor(
              value=false_fail_rate, name="false_fail_rate", dtype=dtype
          )
          tf.debugging.assert_scalar(false_fail_rate)  # Static shape
          itemwise_false_fail_rate = _itemwise_error_rate(
              total_rate=false_fail_rate,
              param_tensors=[],
              samples_tensor=samples,
          )
          n = tf.shape(samples)[0]
          envelope = _dkwm_cdf_envelope(n, itemwise_false_fail_rate)
          distance = kolmogorov_smirnov_distance(
              samples, cdf, left_continuous_cdf
          )
          return tf.debugging.assert_less_equal(
              distance,
              envelope,
              message="Empirical CDF outside K-S envelope",
          )
  
imports: import functools
  import itertools
  import tensorflow.compat.v2
  from tensorflow_probability.python.internal import distribution_util
  from tensorflow_probability.python.internal import dtype_util
  from tensorflow_probability.python.internal import prefer_static
  from tensorflow_probability.python.internal import tensorshape_util
  from tensorflow_probability.python.util.seed_stream import SeedStream
label_desc: Use image node_id 1 for calling a global function with example usage: assert_true_cdf_equal_by_dkwm(samples, cdf, left_continuous_cdf, false_fail_rate, name) and returns: tf