def testGetPluginAsset(self):
   epa = plugin_asset.get_plugin_asset(_ExamplePluginAsset)
   self.assertIsInstance(epa, _ExamplePluginAsset)
   epa2 = plugin_asset.get_plugin_asset(_ExamplePluginAsset)
   self.assertIs(epa, epa2)
   opa = plugin_asset.get_plugin_asset(_OtherExampleAsset)
   self.assertIsNot(epa, opa)

  def testRespectsGraphArgument(self):
    g1 = ops.Graph()
    g2 = ops.Graph()
    e1 = plugin_asset.get_plugin_asset(_ExamplePluginAsset, g1)
    e2 = plugin_asset.get_plugin_asset(_ExamplePluginAsset, g2)

    self.assertEqual(e1, plugin_asset.get_all_plugin_assets(g1)[0])
    self.assertEqual(e2, plugin_asset.get_all_plugin_assets(g2)[0])

  def testEndpointsNoAssets(self):
    g = ops.Graph()
    with g.as_default():
      plugin_asset.get_plugin_asset(projector_plugin.ProjectorPluginAsset)

    fw = writer.FileWriter(self.log_dir, graph=g)
    fw.close()

    self._SetupWSGIApp()
    run_json = self._GetJson('/data/plugin/projector/runs')
    self.assertEqual(run_json, [])

 def testSimplePluginCase(self):
   tempdir = self.get_temp_dir()
   with ops.Graph().as_default() as g:
     plugin_asset.get_plugin_asset(PluginAlpha)
     fw = writer.FileWriter(tempdir)
     fw.add_graph(g)
   self.assertEqual(["Alpha"], plugin_asset_util.ListPlugins(tempdir))
   assets = plugin_asset_util.ListAssets(tempdir, "Alpha")
   self.assertEqual(["contents.txt"], assets)
   contents = plugin_asset_util.RetrieveAsset(tempdir, "Alpha", "contents.txt")
   self.assertEqual("hello world", contents)

  def testNoAssetsProperSerializationOnDisk(self):
    logdir = self.get_temp_dir()
    plugin_dir = os.path.join(logdir, writer._PLUGINS_DIR,
                              projector_plugin.ProjectorPluginAsset.plugin_name)

    with ops.Graph().as_default() as g:
      plugin_asset.get_plugin_asset(projector_plugin.ProjectorPluginAsset)
      fw = writer.FileWriter(logdir)
      fw.add_graph(g)

    with gfile.Open(os.path.join(plugin_dir, 'projector_config.pbtxt')) as f:
      content = f.read()
    self.assertEqual(content, '')

  def testAddEmbeddingThumbnailListEntriesNot3DTensors(self):
    manager = plugin_asset.get_plugin_asset(
        projector_plugin.ProjectorPluginAsset)

    with self.assertRaises(ValueError):
      manager.add_embedding('test3', np.array([[1]]), thumbnails=[[1, 2, 3]],
                            thumbnail_dim=[1, 1])

  def testEndpointsMetadataForVariableAssets(self):
    self._GenerateProjectorTestData()
    g = ops.Graph()
    with g.as_default():
      manager = plugin_asset.get_plugin_asset(
          projector_plugin.ProjectorPluginAsset)

    metadata = projector_plugin.EmbeddingMetadata(3)
    metadata.add_column('labels', ['a', 'b', 'c'])
    manager.add_metadata_for_embedding_variable('test', metadata)

    fw = writer.FileWriter(self.log_dir, graph=g)
    fw.close()

    self._SetupWSGIApp()
    run_json = self._GetJson('/data/plugin/projector/runs')
    self.assertTrue(run_json)

    run = run_json[0]
    metedata_query = '/data/plugin/projector/metadata?run=%s&name=test' % run
    metadata_tsv = self._Get(metedata_query).data
    self.assertEqual(metadata_tsv, b'a\nb\nc\n')

    unk_tensor_query = '/data/plugin/projector/tensor?run=%s&name=test' % run
    response = self._Get(unk_tensor_query)
    self.assertEqual(response.status_code, 400)

    expected_tensor = np.array([[6, 6]], dtype=np.float32)
    tensor_query = '/data/plugin/projector/tensor?run=%s&name=var1' % run
    tensor_bytes = self._Get(tensor_query).data
    self._AssertTensorResponse(tensor_bytes, expected_tensor)

def text_summary(name, tensor, collections=None):
  """Summarizes textual data.

  Text data summarized via this plugin will be visible in the Text Dashboard
  in TensorBoard.

  Args:
    name: A name for the generated node. Will also serve as a series name in
      TensorBoard.
    tensor: a scalar string-type Tensor to summarize.
    collections: Optional list of ops.GraphKeys.  The collections to add the
      summary to.  Defaults to [_ops.GraphKeys.SUMMARIES]

  Returns:
    A  TensorSummary op that is configured so that TensorBoard will recognize
    that it contains textual data. The TensorSummary is a scalar `Tensor` of
    type `string` which contains `Summary` protobufs.

  Raises:
    ValueError: If tensor has the wrong shape or type.
  """
  if tensor.dtype != dtypes.string:
    raise ValueError("Expected tensor %s to have dtype string, got %s" %
                     (tensor.name, tensor.dtype))

  if tensor.shape.ndims != 0:
    raise ValueError("Expected tensor %s to be scalar, has shape %s" %
                     (tensor.name, tensor.shape))

  t_summary = tensor_summary(name, tensor, collections=collections)
  text_assets = plugin_asset.get_plugin_asset(TextSummaryPluginAsset)
  text_assets.register_tensor(t_summary.op.name)
  return t_summary

  def testEndpointsComboTensorAssetsAndCheckpoint(self):
    self._GenerateProjectorTestData()
    g = ops.Graph()
    with g.as_default():
      manager = plugin_asset.get_plugin_asset(
          projector_plugin.ProjectorPluginAsset)

    metadata = projector_plugin.EmbeddingMetadata(3)
    metadata.add_column('labels', ['a', 'b', 'c'])
    manager.add_metadata_for_embedding_variable('var1', metadata)

    new_tensor_values = np.array([[1, 2], [3, 4], [5, 6]])
    manager.add_embedding('new_tensor', new_tensor_values)

    fw = writer.FileWriter(self.log_dir, graph=g)
    fw.close()

    self._SetupWSGIApp()
    run_json = self._GetJson('/data/plugin/projector/runs')
    self.assertTrue(run_json)

    run = run_json[0]
    var1_values = np.array([[6, 6]], dtype=np.float32)
    var1_tensor_query = '/data/plugin/projector/tensor?run=%s&name=var1' % run
    tensor_bytes = self._Get(var1_tensor_query).data
    self._AssertTensorResponse(tensor_bytes, var1_values)

    metadata_query = '/data/plugin/projector/metadata?run=%s&name=var1' % run
    metadata_tsv = self._Get(metadata_query).data
    self.assertEqual(metadata_tsv, b'a\nb\nc\n')

    tensor_query = '/data/plugin/projector/tensor?run=%s&name=new_tensor' % run
    tensor_bytes = self._Get(tensor_query).data
    self._AssertTensorResponse(tensor_bytes, new_tensor_values)

  def testAddEmbeddingThumbnailListNotOfRank4(self):
    manager = plugin_asset.get_plugin_asset(
        projector_plugin.ProjectorPluginAsset)

    with self.assertRaises(ValueError):
      manager.add_embedding('test2', np.array([[1]]),
                            thumbnails=np.array([[1]]), thumbnail_dim=[1, 1])

def text_summary(name, tensor, collections=None):
  """Summarizes textual data.

  Text data summarized via this plugin will be visible in the Text Dashboard
  in TensorBoard. The standard TensorBoard Text Dashboard will render markdown
  in the strings, and will automatically organize 1d and 2d tensors into tables.
  If a tensor with more than 2 dimensions is provided, a 2d subarray will be
  displayed along with a warning message. (Note that this behavior is not
  intrinsic to the text summary api, but rather to the default TensorBoard text
  plugin.)

  Args:
    name: A name for the generated node. Will also serve as a series name in
      TensorBoard.
    tensor: a string-type Tensor to summarize.
    collections: Optional list of ops.GraphKeys.  The collections to add the
      summary to.  Defaults to [_ops.GraphKeys.SUMMARIES]

  Returns:
    A  TensorSummary op that is configured so that TensorBoard will recognize
    that it contains textual data. The TensorSummary is a scalar `Tensor` of
    type `string` which contains `Summary` protobufs.

  Raises:
    ValueError: If tensor has the wrong type.
  """
  if tensor.dtype != dtypes.string:
    raise ValueError("Expected tensor %s to have dtype string, got %s" %
                     (tensor.name, tensor.dtype))

  t_summary = tensor_summary(name, tensor, collections=collections)
  text_assets = plugin_asset.get_plugin_asset(TextSummaryPluginAsset)
  text_assets.register_tensor(t_summary.op.name)
  return t_summary

  def testPluginAssetSerialized(self):
    with ops.Graph().as_default() as g:
      plugin_asset.get_plugin_asset(ExamplePluginAsset)

      logdir = self.get_temp_dir()
      fw = writer.FileWriter(logdir)
      fw.add_graph(g)
    plugin_dir = os.path.join(logdir, writer._PLUGINS_DIR, "example")

    with gfile.Open(os.path.join(plugin_dir, "foo.txt"), "r") as f:
      content = f.read()
    self.assertEqual(content, "foo!")

    with gfile.Open(os.path.join(plugin_dir, "bar.txt"), "r") as f:
      content = f.read()
    self.assertEqual(content, "bar!")

  def testAddEmbeddingWithSpriteImageButNoThumbnailDim(self):
    manager = plugin_asset.get_plugin_asset(
        projector_plugin.ProjectorPluginAsset)

    thumbnails = np.array([[[1, 2, 3], [4, 5, 6], [7, 8, 9]]])
    with self.assertRaises(ValueError):
      manager.add_embedding(
          'test', np.array([[1], [2], [3]]), thumbnails=thumbnails)

  def testAddEmbeddingThumbnailDimNotOfLength2(self):
    manager = plugin_asset.get_plugin_asset(
        projector_plugin.ProjectorPluginAsset)

    thumbnails = np.array([[[1, 2, 3], [4, 5, 6], [7, 8, 9]]])
    with self.assertRaises(ValueError):
      manager.add_embedding(
          'test', np.array([[1], [2], [3]]), thumbnails=thumbnails,
          thumbnail_dim=[4])

  def testAddEmbeddingWithMetadataOfIncorrectLength(self):
    manager = plugin_asset.get_plugin_asset(
        projector_plugin.ProjectorPluginAsset)

    metadata = projector_plugin.EmbeddingMetadata(3)
    metadata.add_column('labels', ['a', 'b', 'c'])
    # values has length 2, while metadata has length 3.
    values = np.array([[1], [2]])

    with self.assertRaises(ValueError):
      manager.add_embedding('test', values, metadata)

  def testEndpointsTensorAndMetadataAssets(self):
    g = ops.Graph()
    with g.as_default():
      manager = plugin_asset.get_plugin_asset(
          projector_plugin.ProjectorPluginAsset)

    metadata = projector_plugin.EmbeddingMetadata(3)
    metadata.add_column('labels', ['a', 'b', 'c'])
    manager.add_metadata_for_embedding_variable('test', metadata)
    expected_tensor = np.array([[1, 2], [3, 4], [5, 6]])
    image1 = np.array([[[1, 2, 3], [4, 5, 6]],
                       [[7, 8, 9], [10, 11, 12]]])
    image2 = np.array([[[10, 20, 30], [40, 50, 60]],
                       [[70, 80, 90], [100, 110, 120]]])
    manager.add_embedding('emb', expected_tensor, metadata, [image1, image2],
                          [2, 2])

    fw = writer.FileWriter(self.log_dir, graph=g)
    fw.close()

    self._SetupWSGIApp()
    run_json = self._GetJson('/data/plugin/projector/runs')
    self.assertTrue(run_json)

    run = run_json[0]
    metadata_query = '/data/plugin/projector/metadata?run=%s&name=emb' % run
    metadata_tsv = self._Get(metadata_query).data
    self.assertEqual(metadata_tsv, b'a\nb\nc\n')

    unk_metadata_query = '/data/plugin/projector/metadata?run=%s&name=q' % run
    response = self._Get(unk_metadata_query)
    self.assertEqual(response.status_code, 400)

    tensor_query = '/data/plugin/projector/tensor?run=%s&name=emb' % run
    tensor_bytes = self._Get(tensor_query).data
    self._AssertTensorResponse(tensor_bytes, expected_tensor)

    unk_tensor_query = '/data/plugin/projector/tensor?run=%s&name=var1' % run
    response = self._Get(unk_tensor_query)
    self.assertEqual(response.status_code, 400)

    image_query = '/data/plugin/projector/sprite_image?run=%s&name=emb' % run
    image_bytes = self._Get(image_query).data
    with ops.Graph().as_default():
      s = session.Session()
      image_array = image_ops.decode_png(image_bytes).eval(session=s).tolist()
    expected_sprite_image = [
        [[1, 2, 3], [4, 5, 6], [10, 20, 30], [40, 50, 60]],
        [[7, 8, 9], [10, 11, 12], [70, 80, 90], [100, 110, 120]],
        [[0, 0, 0], [0, 0, 0], [0, 0, 0], [0, 0, 0]],
        [[0, 0, 0], [0, 0, 0], [0, 0, 0], [0, 0, 0]]
    ]
    self.assertEqual(image_array, expected_sprite_image)

  def testEventMultiplexerIntegration(self):
    tempdir = self.get_temp_dir()
    with ops.Graph().as_default() as g:
      plugin_instance = plugin_asset.get_plugin_asset(PluginAlpha)
      plugin_instance.contents = "graph one"
      plugin_asset.get_plugin_asset(PluginBeta)

      fw = writer.FileWriter(os.path.join(tempdir, "one"))
      fw.add_graph(g)
      fw.close()

    with ops.Graph().as_default() as g:
      plugin_instance = plugin_asset.get_plugin_asset(PluginAlpha)
      plugin_instance.contents = "graph two"
      fw = writer.FileWriter(os.path.join(tempdir, "two"))
      fw.add_graph(g)
      fw.close()

    multiplexer = event_multiplexer.EventMultiplexer()
    multiplexer.AddRunsFromDirectory(tempdir)

    self.assertEqual(
        multiplexer.PluginAssets("Alpha"),
        {"one": ["contents.txt"], "two": ["contents.txt"]})
    self.assertEqual(
        multiplexer.RetrievePluginAsset("one", "Alpha", "contents.txt"),
        "graph one")
    self.assertEqual(
        multiplexer.RetrievePluginAsset("one", "Beta", "contents.txt"),
        "hello world")
    self.assertEqual(
        multiplexer.RetrievePluginAsset("two", "Alpha", "contents.txt"),
        "graph two")

    self.assertEqual(
        multiplexer.PluginAssets("Beta"),
        {"one": ["contents.txt"], "two": []})
    self.assertEqual(multiplexer.PluginAssets("Gamma"), {"one": [], "two": []})

  def testAddEmbeddingNoMetadata(self):
    manager = plugin_asset.get_plugin_asset(
        projector_plugin.ProjectorPluginAsset)
    manager.add_embedding('test', np.array([[1, 2, 3.1]]))

    config = projector_config_pb2.ProjectorConfig()
    embedding = config.embeddings.add()
    embedding.tensor_name = 'test'
    embedding.tensor_shape.extend([1, 3])
    embedding.tensor_path = 'test_values.tsv'
    expected_config_pbtxt = text_format.MessageToString(config)

    self.assertEqual(manager.assets(), {
        'projector_config.pbtxt': expected_config_pbtxt,
        'test_values.tsv': b'1\t2\t3.1\n'
    })

  def testEndpointsMetadataForVariableAssetsButNoCheckpoint(self):
    g = ops.Graph()
    with g.as_default():
      manager = plugin_asset.get_plugin_asset(
          projector_plugin.ProjectorPluginAsset)

    metadata = projector_plugin.EmbeddingMetadata(3)
    metadata.add_column('labels', ['a', 'b', 'c'])
    manager.add_metadata_for_embedding_variable('test', metadata)

    fw = writer.FileWriter(self.log_dir, graph=g)
    fw.close()

    self._SetupWSGIApp()
    run_json = self._GetJson('/data/plugin/projector/runs')
    self.assertEqual(run_json, [])

  def testAddMetadataForVariable(self):
    manager = plugin_asset.get_plugin_asset(
        projector_plugin.ProjectorPluginAsset)
    metadata = projector_plugin.EmbeddingMetadata(3)
    metadata.add_column('Labels', ['a', 'b', 'c'])
    manager.add_metadata_for_embedding_variable('test', metadata)

    config = projector_config_pb2.ProjectorConfig()
    embedding = config.embeddings.add()
    embedding.tensor_name = 'test'
    embedding.metadata_path = 'test_metadata.tsv'
    expected_config_pbtxt = text_format.MessageToString(config)

    self.assertEqual(manager.assets(), {
        'projector_config.pbtxt': expected_config_pbtxt,
        'test_metadata.tsv': 'a\nb\nc\n'
    })