class_docstr: Extracts Score2Perf examples from NoteSequence protos.
class_docstr_tok: ["Extracts","Score2Perf","examples","from","NoteSequence","protos","."]
code: def process(self, kv):
      # Seed random number generator based on key so that hop times are
      # deterministic.
      key, ns_str = kv
      m = hashlib.md5(key.encode("utf-8"))
      random.seed(int(m.hexdigest(), 16))
  
      # Deserialize NoteSequence proto.
      ns = note_seq.NoteSequence.FromString(ns_str)
  
      # Apply sustain pedal.
      ns = sequences_lib.apply_sustain_control_changes(ns)
  
      # Remove control changes as there are potentially a lot of them and they are
      # no longer needed.
      del ns.control_changes[:]
  
      if (
          self._min_hop_size_seconds
          and ns.total_time < self._min_hop_size_seconds
      ):
          Metrics.counter(
              "extract_examples", "sequence_too_short"
          ).inc()
          return
  
      sequences = []
      for _ in range(self._num_replications):
          if self._max_hop_size_seconds:
              if (
                  self._max_hop_size_seconds
                  == self._min_hop_size_seconds
              ):
                  # Split using fixed hop size.
                  sequences += sequences_lib.split_note_sequence(
                      ns, self._max_hop_size_seconds
                  )
              else:
                  # Sample random hop positions such that each segment size is within
                  # the specified range.
                  hop_times = [0.0]
                  while (
                      hop_times[-1]
                      <= ns.total_time - self._min_hop_size_seconds
                  ):
                      if (
                          hop_times[-1] + self._max_hop_size_seconds
                          < ns.total_time
                      ):
                          # It's important that we get a valid hop size here, since the
                          # remainder of the sequence is too long.
                          max_offset = min(
                              self._max_hop_size_seconds,
                              ns.total_time
                              - self._min_hop_size_seconds
                              - hop_times[-1],
                          )
                      else:
                          # It's okay if the next hop time is invalid (in which case we'll
                          # just stop).
                          max_offset = self._max_hop_size_seconds
                      offset = random.uniform(
                          self._min_hop_size_seconds, max_offset
                      )
                      hop_times.append(hop_times[-1] + offset)
                  # Split at the chosen hop times (ignoring zero and the final invalid
                  # time).
                  sequences += sequences_lib.split_note_sequence(
                      ns, hop_times[1:-1]
                  )
          else:
              sequences += [ns]
  
      for performance_sequence in sequences:
          if self._encode_score_fns:
              # We need to extract a score.
              if not self._absolute_timing:
                  # Beats are required to extract a score with metric timing.
                  beats = [
                      ta
                      for ta in performance_sequence.text_annotations
                      if ta.annotation_type == BEAT
                      and ta.time <= performance_sequence.total_time
                  ]
                  if len(beats) < 2:
                      Metrics.counter(
                          "extract_examples", "not_enough_beats"
                      ).inc()
                      continue
  
                  # Ensure the sequence starts and ends on a beat.
                  performance_sequence = (
                      sequences_lib.extract_subsequence(
                          performance_sequence,
                          start_time=min(beat.time for beat in beats),
                          end_time=max(beat.time for beat in beats),
                      )
                  )
  
                  # Infer beat-aligned chords (only for relative timing).
                  try:
                      chord_inference.infer_chords_for_sequence(
                          performance_sequence,
                          chord_change_prob=0.25,
                          chord_note_concentration=50.0,
                          add_key_signatures=True,
                      )
                  except chord_inference.ChordInferenceError:
                      Metrics.counter(
                          "extract_examples", "chord_inference_failed"
                      ).inc()
                      continue
  
              # Infer melody regardless of relative/absolute timing.
              try:
                  melody_instrument = (
                      melody_inference.infer_melody_for_sequence(
                          performance_sequence,
                          melody_interval_scale=2.0,
                          rest_prob=0.1,
                          instantaneous_non_max_pitch_prob=1e-15,
                          instantaneous_non_empty_rest_prob=0.0,
                          instantaneous_missing_pitch_prob=1e-15,
                      )
                  )
              except melody_inference.MelodyInferenceError:
                  Metrics.counter(
                      "extract_examples", "melody_inference_failed"
                  ).inc()
                  continue
  
              if not self._absolute_timing:
                  # Now rectify detected beats to occur at fixed tempo.
                  # TODO(iansimon): also include the alignment
                  (
                      score_sequence,
                      unused_alignment,
                  ) = sequences_lib.rectify_beats(
                      performance_sequence, beats_per_minute=SCORE_BPM
                  )
              else:
                  # Score uses same timing as performance.
                  score_sequence = copy.deepcopy(performance_sequence)
  
              # Remove melody notes from performance.
              performance_notes = []
              for note in performance_sequence.notes:
                  if note.instrument != melody_instrument:
                      performance_notes.append(note)
              del performance_sequence.notes[:]
              performance_sequence.notes.extend(performance_notes)
  
              # Remove non-melody notes from score.
              score_notes = []
              for note in score_sequence.notes:
                  if note.instrument == melody_instrument:
                      score_notes.append(note)
              del score_sequence.notes[:]
              score_sequence.notes.extend(score_notes)
  
              # Remove key signatures and beat/chord annotations from performance.
              del performance_sequence.key_signatures[:]
              del performance_sequence.text_annotations[:]
  
              Metrics.counter(
                  "extract_examples", "extracted_score"
              ).inc()
  
          for augment_fn in self._augment_fns:
              # Augment and encode the performance.
              try:
                  augmented_performance_sequence = augment_fn(
                      performance_sequence
                  )
              except DataAugmentationError:
                  Metrics.counter(
                      "extract_examples", "augment_performance_failed"
                  ).inc()
                  continue
              example_dict = {
                  "targets": self._encode_performance_fn(
                      augmented_performance_sequence
                  )
              }
              if not example_dict["targets"]:
                  Metrics.counter(
                      "extract_examples", "skipped_empty_targets"
                  ).inc()
                  continue
  
              if (
                  self._random_crop_length
                  and len(example_dict["targets"])
                  > self._random_crop_length
              ):
                  # Take a random crop of the encoded performance.
                  max_offset = (
                      len(example_dict["targets"])
                      - self._random_crop_length
                  )
                  offset = random.randrange(max_offset + 1)
                  example_dict["targets"] = example_dict["targets"][
                      offset : offset + self._random_crop_length
                  ]
  
              if self._encode_score_fns:
                  # Augment the extracted score.
                  try:
                      augmented_score_sequence = augment_fn(
                          score_sequence
                      )
                  except DataAugmentationError:
                      Metrics.counter(
                          "extract_examples", "augment_score_failed"
                      ).inc()
                      continue
  
                  # Apply all score encoding functions.
                  skip = False
                  for (
                      name,
                      encode_score_fn,
                  ) in self._encode_score_fns.items():
                      example_dict[name] = encode_score_fn(
                          augmented_score_sequence
                      )
                      if not example_dict[name]:
                          Metrics.counter(
                              "extract_examples",
                              "skipped_empty_%s" % name,
                          ).inc()
                          skip = True
                          break
                  if skip:
                      continue
  
              Metrics.counter(
                  "extract_examples", "encoded_example"
              ).inc()
              Metrics.distribution(
                  "extract_examples", "performance_length_in_seconds"
              ).update(int(augmented_performance_sequence.total_time))
  
              yield generator_utils.to_example(example_dict)
  
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 2 for calling the ExtractExamplesDoFn obj's underlying member method code with example usage: obj.process(kv) without return types