code: def generate_examples(
      input_transform,
      output_dir,
      problem_name,
      splits,
      min_hop_size_seconds,
      max_hop_size_seconds,
      num_replications,
      min_pitch,
      max_pitch,
      encode_performance_fn,
      encode_score_fns=None,
      augment_fns=None,
      absolute_timing=False,
      random_crop_length=None,
  ):
      """Generate data for a Score2Perf problem.
  
      Args:
        input_transform: The input PTransform object that reads input NoteSequence
            protos, or dictionary mapping split names to such PTransform objects.
            Should produce `(id, NoteSequence)` tuples.
        output_dir: The directory to write the resulting TFRecord file containing
            examples.
        problem_name: Name of the Tensor2Tensor problem, used as a base filename
            for generated data.
        splits: A dictionary of split names and their probabilities. Probabilites
            should add up to 1. If `input_filename` is a dictionary, this argument
            will be ignored.
        min_hop_size_seconds: Minimum hop size in seconds at which input
            NoteSequence protos can be split. Can also be a dictionary mapping split
            name to minimum hop size.
        max_hop_size_seconds: Maximum hop size in seconds at which input
            NoteSequence protos can be split. If zero or None, will not split at
            all. Can also be a dictionary mapping split name to maximum hop size.
        num_replications: Number of times input NoteSequence protos will be
            replicated prior to splitting.
        min_pitch: Minimum MIDI pitch value; notes with lower pitch will be dropped.
        max_pitch: Maximum MIDI pitch value; notes with greater pitch will be
            dropped.
        encode_performance_fn: Required performance encoding function.
        encode_score_fns: Optional dictionary of named score encoding functions.
        augment_fns: Optional list of data augmentation functions. Only applied in
            the 'train' split.
        absolute_timing: If True, each score will use absolute instead of tempo-
            relative timing. Since chord inference depends on having beats, the
            score will only contain melody.
        random_crop_length: If specified, crop each encoded performance to this
            length. Cannot be specified if using scores.
  
      Raises:
        ValueError: If split probabilities do not add up to 1, or if splits are not
            provided but `input_filename` is not a dictionary.
      """
      # Make sure Beam's log messages are not filtered.
      logging.getLogger().setLevel(logging.INFO)
  
      if isinstance(input_transform, dict):
          split_names = input_transform.keys()
      else:
          if not splits:
              raise ValueError(
                  "Split probabilities must be provided if input is not presplit."
              )
          split_names, split_probabilities = zip(*splits.items())
          cumulative_splits = list(
              zip(split_names, np.cumsum(split_probabilities))
          )
          if cumulative_splits[-1][1] != 1.0:
              raise ValueError(
                  "Split probabilities must sum to 1; got %f"
                  % cumulative_splits[-1][1]
              )
  
      # Check for existence of prior outputs. Since the number of shards may be
      # different, the prior outputs will not necessarily be overwritten and must
      # be deleted explicitly.
      output_filenames = [
          os.path.join(
              output_dir, "%s-%s.tfrecord" % (problem_name, split_name)
          )
          for split_name in split_names
      ]
      for split_name, output_filename in zip(
          split_names, output_filenames
      ):
          existing_output_filenames = tf.gfile.Glob(
              output_filename + "*"
          )
          if existing_output_filenames:
              tf.logging.info(
                  "Data files already exist for split %s in problem %s, deleting.",
                  split_name,
                  problem_name,
              )
              for filename in existing_output_filenames:
                  tf.gfile.Remove(filename)
  
      pipeline_options = beam.options.pipeline_options.PipelineOptions(
          FLAGS.pipeline_options.split(",")
      )
  
      with beam.Pipeline(options=pipeline_options) as p:
          if isinstance(input_transform, dict):
              # Input data is already partitioned into splits.
              split_partitions = [
                  p
                  | "input_transform_%s" % split_name
                  >> input_transform[split_name]
                  for split_name in split_names
              ]
          else:
              # Read using a single PTransform.
              p |= "input_transform" >> input_transform
              split_partitions = p | "partition" >> beam.Partition(
                  functools.partial(select_split, cumulative_splits),
                  len(cumulative_splits),
              )
  
          for split_name, output_filename, s in zip(
              split_names, output_filenames, split_partitions
          ):
              if isinstance(min_hop_size_seconds, dict):
                  min_hop = min_hop_size_seconds[split_name]
              else:
                  min_hop = min_hop_size_seconds
              if isinstance(max_hop_size_seconds, dict):
                  max_hop = max_hop_size_seconds[split_name]
              else:
                  max_hop = max_hop_size_seconds
              s |= "preshuffle_%s" % split_name >> beam.Reshuffle()
              s |= "filter_invalid_notes_%s" % split_name >> beam.Map(
                  functools.partial(
                      filter_invalid_notes, min_pitch, max_pitch
                  )
              )
              s |= "extract_examples_%s" % split_name >> beam.ParDo(
                  ExtractExamplesDoFn(
                      min_hop,
                      max_hop,
                      num_replications if split_name == "train" else 1,
                      encode_performance_fn,
                      encode_score_fns,
                      augment_fns if split_name == "train" else None,
                      absolute_timing,
                      random_crop_length,
                  )
              )
              s |= "shuffle_%s" % split_name >> beam.Reshuffle()
              s |= "write_%s" % split_name >> beam.io.WriteToTFRecord(
                  output_filename,
                  coder=beam.coders.ProtoCoder(tf.train.Example),
              )
  
imports: import copy
  import functools
  import hashlib
  import logging
  import os
  import random
  import typing
  import apache_beam
  from apache_beam import typehints
  from apache_beam.metrics import Metrics
  from magenta.models.score2perf import music_encoders
  import note_seq
  from note_seq import chord_inference
  from note_seq import melody_inference
  from note_seq import sequences_lib
  import numpy
  from tensor2tensor.data_generators import generator_utils
  import tensorflow.compat.v1
label_desc: Use image node_id 3 for calling a global function with example usage: generate_examples(input_transform, output_dir, problem_name, splits, min_hop_size_seconds, max_hop_size_seconds, num_replications, min_pitch, max_pitch, encode_performance_fn, encode_score_fns, augment_fns, absolute_timing, random_crop_length) without return types