audio_plugin_test.py 文件源码

def setUp(self):
    self.log_dir = tempfile.mkdtemp()

    # We use numpy.random to generate audio. We seed to avoid non-determinism
    # in this test.
    numpy.random.seed(42)

    # Create old-style audio summaries for run "foo".
    tf.reset_default_graph()
    sess = tf.Session()
    placeholder = tf.placeholder(tf.float32)
    tf.summary.audio(name="baz", tensor=placeholder, sample_rate=44100)
    merged_summary_op = tf.summary.merge_all()
    foo_directory = os.path.join(self.log_dir, "foo")
    writer = tf.summary.FileWriter(foo_directory)
    writer.add_graph(sess.graph)
    for step in xrange(2):
      # The floats (sample data) range from -1 to 1.
      writer.add_summary(sess.run(merged_summary_op, feed_dict={
          placeholder: numpy.random.rand(42, 22050) * 2 - 1
      }), global_step=step)
    writer.close()

    # Create new-style audio summaries for run "bar".
    tf.reset_default_graph()
    sess = tf.Session()
    audio_placeholder = tf.placeholder(tf.float32)
    labels_placeholder = tf.placeholder(tf.string)
    summary.op("quux", audio_placeholder, sample_rate=44100,
               labels=labels_placeholder,
               description="how do you pronounce that, anyway?")
    merged_summary_op = tf.summary.merge_all()
    bar_directory = os.path.join(self.log_dir, "bar")
    writer = tf.summary.FileWriter(bar_directory)
    writer.add_graph(sess.graph)
    for step in xrange(2):
      # The floats (sample data) range from -1 to 1.
      writer.add_summary(sess.run(merged_summary_op, feed_dict={
          audio_placeholder: numpy.random.rand(42, 11025, 1) * 2 - 1,
          labels_placeholder: [
              tf.compat.as_bytes('step **%s**, sample %s' % (step, sample))
              for sample in xrange(42)
          ],
      }), global_step=step)
    writer.close()

    # Start a server with the plugin.
    multiplexer = event_multiplexer.EventMultiplexer({
        "foo": foo_directory,
        "bar": bar_directory,
    })
    context = base_plugin.TBContext(
        logdir=self.log_dir, multiplexer=multiplexer)
    self.plugin = audio_plugin.AudioPlugin(context)
    wsgi_app = application.TensorBoardWSGIApp(
        self.log_dir, [self.plugin], multiplexer, reload_interval=0,
        path_prefix='')
    self.server = werkzeug_test.Client(wsgi_app, wrappers.BaseResponse)