code: def sequential_rolling_range(
      artifacts,
      *,
      start_span_number: Optional[int] = None,
      num_spans: int = 1,
      skip_num_recent_spans: int = 0,
      keep_all_versions: bool = False,
      exclude_span_numbers: Sequence[int] = (),
  ):
      """Returns artifacts with spans in a sequential rolling range.
  
      Sequential rolling range is a sliding window on the oldest consecutive spans.
  
      The consecutive spans must be in the range:
      [start_span_number, max_span - skip_num_recent_spans], where max_span is the
      maximum span present in the artifacts. This range is modified to account for
      exclude_span_numbers, for details see the ConsecutiveSpans ResolverOp
      implementation.
  
      The window size is num_spans and has a stride of 1. If the spans are not
      consecutive, then the sequential rolling range waits for the missing span to
      arrive.
  
      This resolver function is based on the span-version semantics, which only
      considers the latest version of each span. If you want to keep all versions,
      then set keep_all_versions=True. Input artifacts must have both "span" int
      property and "version" int property.
  
      Corresponds to SequentialRollingRange in TFX.
  
      Example usage:
  
        Consider 5 artifacts [A, B, C, D, E] with spans = [1, 2, 3, 4, 7].
  
        sequential_rolling_range(
            start_span_number=1,
            num_spans=3,
            skip_num_recent_spans=1,
            keep_all_versions=False,
            exclude_span_numbers=[])
  
        The consecutive spans to consider are [1, 2, 3, 4]
  
        The artifacts will be returned with a sliding window of size num_spans=3 and
        stride 1 applied:
  
        [[A, B, C], [B, C, D]]
  
        However, if nums_spans=5, there are only 4 consecutive spans to consider, so
        [], no artifacts, will be returned.
  
      Since sequential_rolling_range returns multiple windows, it must be used
      together with ForEach. For example:
  
        with ForEach(sequential_rolling_range(
            all_examples, num_spans=10)) as examples_window:
          trainer = Trainer(examples=shuffle(examples_window))
  
      Args:
        artifacts: The artifacts to filter.
        start_span_number: The smallest span number to keep, inclusive. Optional, if
          not set then defaults to the minimum span. If the start_span_number is
          configured wrong (so that it is smaller than the first span number), we
          will wait indefinitely until the missing spans between start_span_number
          and the first span number to be appeared.
        num_spans: The length of the range. If num_spans <= 0, then num_spans is set
          to the total number of artifacts with consecutive spans in the range. Note
          that this is also the size of the sliding window of the sequential rolling
          range.
        skip_num_recent_spans: Number of most recently available (largest) spans to
          skip. Defaults to 0.
        keep_all_versions: If true, all artifacts with spans in the range are kept.
          If false then if multiple artifacts have the same span, only the span with
          the latest version is kept. Defaults to False.
        exclude_span_numbers: The list of missing/bad span numbers to exclude.
  
      Returns:
        Artifacts with spans in the sequential rolling range.
      """
      resolved_artifacts = ops.ConsecutiveSpans(
          artifacts,
          first_span=start_span_number
          if start_span_number is not None
          else -1,
          skip_last_n=skip_num_recent_spans,
          keep_all_versions=keep_all_versions,
          denylist=exclude_span_numbers,
      )
  
      return ops.SlidingWindow(
          resolved_artifacts, window_size=num_spans
      )
  
imports: from typing import Optional, Sequence, Union
  from absl import logging
  from tfx.dsl.input_resolution import resolver_function
  from tfx.dsl.input_resolution.ops import ops
  from tfx.types import artifact
  from tfx.types import channel
label_desc: Use image node_id 9 for calling a global function with example usage: sequential_rolling_range(artifacts) and returns: ops