code: def get_minibatch(
      bmuf,
      working_dir,
      mb_source,
      num_data_partitions=1,
      partition_index=0,
  ):
      from cntk.io import (
          MinibatchSource,
          CTFDeserializer,
          StreamDef,
          StreamDefs,
      )
  
      if mb_source == "numpy":
          assert (
              num_data_partitions == 1
          )  # numpy option does not support more than one partition in this impl.
          assert partition_index == 0
          for i in range(num_batches):
              features = []
              labels = []
              for j in range(batch_size):
                  seq_len_j = [seq_len, seq_len + 5, seq_len - 5][j % 3]
                  x = np.random.rand(seq_len_j, feat_dim).astype(
                      np.float32
                  )
                  y = np.random.rand(seq_len_j, label_dim).astype(
                      np.float32
                  )
                  features.append(x)
                  labels.append(y)
              yield {bmuf.feat: features, bmuf.label: labels}
  
      if mb_source in ("ctf_utterance", "ctf_frame", "ctf_bptt"):
          if mb_source == "ctf_frame":
              # frame mode data without sequence ids.
              ctf_data = ctf_data = """\
  |S0 0.49  0.18  0.84  0.7   0.59 |S1 0.12  0.24  0.14
  |S0 0.69  0.63  0.47  0.93  0.69 |S1 0.34  0.85  0.17
  |S0 0.04  0.5   0.39  0.86  0.28 |S1 0.62  0.36  0.53
  |S0 0.71  0.9   0.15  0.83  0.18 |S1 0.2   0.74  0.04
  |S0 0.38  0.67  0.46  0.53  0.75 |S1 0.6   0.14  0.35
  |S0 0.94  0.54  0.09  0.55  0.08 |S1 0.07  0.53  0.47
  |S0 0.11  0.24  0.17  0.72  0.72 |S1 0.9   0.98  0.18
  |S0 0.3   1.    0.34  0.06  0.78 |S1 0.15  0.69  0.63
  |S0 0.69  0.86  0.59  0.49  0.99 |S1 0.13  0.6   0.21
  """
          # sequence mode data with sequence id
          else:
              ctf_data = ctf_data = """\
  0	|S0 0.49  0.18  0.84  0.7   0.59 |S1 0.12  0.24  0.14
  0	|S0 0.69  0.63  0.47  0.93  0.69 |S1 0.34  0.85  0.17
  0	|S0 0.04  0.5   0.39  0.86  0.28 |S1 0.62  0.36  0.53
  0	|S0 0.71  0.9   0.15  0.83  0.18 |S1 0.2   0.74  0.04
  0	|S0 0.38  0.67  0.46  0.53  0.75 |S1 0.6   0.14  0.35
  0	|S0 0.94  0.54  0.09  0.55  0.08 |S1 0.07  0.53  0.47
  0	|S0 0.11  0.24  0.17  0.72  0.72 |S1 0.9   0.98  0.18
  2	|S0 0.3   1.    0.34  0.06  0.78 |S1 0.15  0.69  0.63
  2	|S0 0.69  0.86  0.59  0.49  0.99 |S1 0.13  0.6   0.21
  """
  
          ctf_file = os.path.join(working_dir, "2seqtest.txt")
          with open(ctf_file, "w") as f:
              f.write(ctf_data)
  
          # ctf_utterance model
          frame_mode = False
          truncation_length = 0
  
          if mb_source == "ctf_frame":
              frame_mode = True
          elif mb_source == "ctf_bptt":
              truncation_length = 2
  
          mbs = MinibatchSource(
              CTFDeserializer(
                  ctf_file,
                  StreamDefs(
                      features=StreamDef(
                          field="S0", shape=feat_dim, is_sparse=False
                      ),
                      labels=StreamDef(
                          field="S1", shape=label_dim, is_sparse=False
                      ),
                  ),
              ),
              randomize=False,
              max_samples=batch_size * num_batches,
              frame_mode=frame_mode,
              truncation_length=truncation_length,
          )
  
          for i in range(num_batches):
              minibatch = mbs.next_minibatch(
                  minibatch_size_in_samples=batch_size,
                  input_map={
                      bmuf.feat: mbs.streams.features,
                      bmuf.label: mbs.streams.labels,
                  },
                  num_data_partitions=num_data_partitions,
                  partition_index=partition_index,
              )
              if not minibatch:
                  break
              yield minibatch
  
file_path: CNTK/bindings/python/cntk/learners/tests/bmuf_metrics_aggregation_test.py
imports: import pytest
  import cntk
  import numpy
  import sys, os
  from distributed_learner_test import mpiexec_execute
  import argparse
  import re
  import platform
label_desc: Use image node_id 1 for calling a global function with example usage: get_minibatch(bmuf, working_dir, mb_source, num_data_partitions, partition_index) without return types