Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.mock_debugger_data_server = tf.compat.v1.test.mock.Mock(
debugger_server_lib.DebuggerDataServer
)
self.mock_debugger_data_server_class = tf.compat.v1.test.mock.Mock(
debugger_server_lib.DebuggerDataServer,
return_value=self.mock_debugger_data_server,
)
tf.compat.v1.test.mock.patch.object(
debugger_server_lib,
"DebuggerDataServer",
self.mock_debugger_data_server_class,
).start()
self.context = base_plugin.TBContext(
logdir=self.log_dir, multiplexer=multiplexer
)
self.plugin = debugger_plugin.DebuggerPlugin(self.context)
self.plugin.listen(self.debugger_data_server_grpc_port)
wsgi_app = application.TensorBoardWSGI([self.plugin])
self.server = werkzeug_test.Client(wsgi_app, wrappers.BaseResponse)
# The debugger data server should be started at the correct port.
self.mock_debugger_data_server_class.assert_called_once_with(
self.debugger_data_server_grpc_port, self.log_dir
)
mock_debugger_data_server = self.mock_debugger_data_server
start = (
mock_debugger_data_server.start_the_debugger_data_receiving_server
)
return -1
def _rel_to_abs_asset_path(fpath, config_fpath):
fpath = os.path.expanduser(fpath)
if not os.path.isabs(fpath):
return os.path.join(os.path.dirname(config_fpath), fpath)
return fpath
def _using_tf():
"""Return true if we're not using the fake TF API stub implementation."""
return tf.__version__ != "stub"
class ProjectorPlugin(base_plugin.TBPlugin):
"""Embedding projector."""
plugin_name = _PLUGIN_PREFIX_ROUTE
def __init__(self, context):
"""Instantiates ProjectorPlugin via TensorBoard core.
Args:
context: A base_plugin.TBContext instance.
"""
self.multiplexer = context.multiplexer
self.logdir = context.logdir
self._handlers = None
self.readers = {}
self.run_paths = None
self._configs = {}
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""The TensorBoard Debugger V2 plugin."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from tensorboard.plugins import base_plugin
class DebuggerV2Plugin(base_plugin.TBPlugin):
"""Debugger V2 Plugin for TensorBoard."""
plugin_name = "debugger-v2"
def __init__(self, context):
"""Instantiates Debugger V2 Plugin via TensorBoard core.
Args:
context: A base_plugin.TBContext instance.
"""
super(DebuggerV2Plugin, self).__init__(context)
def get_plugin_apps(self):
# TODO(cais): Add routes as they are implemented.
return {}
def frontend_metadata(self):
return base_plugin.FrontendMetadata(element_name="tf-scalar-dashboard")
from tensorboard import plugin_util
from tensorboard.backend import http_util
from tensorboard.compat import tf
from tensorboard.plugins import base_plugin
from tensorboard.plugins.audio import metadata
from tensorboard.util import tensor_util
_DEFAULT_MIME_TYPE = "application/octet-stream"
_MIME_TYPES = {
metadata.Encoding.Value("WAV"): "audio/wav",
}
class AudioPlugin(base_plugin.TBPlugin):
"""Audio Plugin for TensorBoard."""
plugin_name = metadata.PLUGIN_NAME
def __init__(self, context):
"""Instantiates AudioPlugin via TensorBoard core.
Args:
context: A base_plugin.TBContext instance.
"""
self._multiplexer = context.multiplexer
def get_plugin_apps(self):
return {
"/audio": self._serve_audio_metadata,
"/individualAudio": self._serve_individual_audio,
import six
from werkzeug import wrappers
import werkzeug
from tensorboard import errors
from tensorboard import plugin_util
from tensorboard.backend import http_util
from tensorboard.plugins import base_plugin
from tensorboard.util import tensor_util
from tensorboard.plugins.scalar import metadata
_SCALAR_PLUGIN_NAME = metadata.PLUGIN_NAME
_PLUGIN_DIRECTORY_PATH_PART = "/data/plugin/example_raw_scalars/"
class ExampleRawScalarsPlugin(base_plugin.TBPlugin):
"""Raw summary example plugin for TensorBoard."""
plugin_name = "example_raw_scalars"
def __init__(self, context):
"""Instantiates ExampleRawScalarsPlugin.
Args:
context: A base_plugin.TBContext instance.
"""
self._multiplexer = context.multiplexer
def get_plugin_apps(self):
return {
"/scalars": self.scalars_route,
"/tags": self._serve_tags,
def frontend_metadata(self):
return base_plugin.FrontendMetadata(element_name="tf-text-dashboard")
def frontend_metadata(self):
return base_plugin.FrontendMetadata(es_module_path="/index.js")
def get_plugin_apps(self):
return {
"/debugger_grpc_host_port": self._serve_debugger_grpc_host_port,
}
@wrappers.Request.application
def _serve_debugger_grpc_host_port(self, request):
# Respond with a -1 port number to indicate the debugger plugin is
# inactive.
return http_util.Respond(
request, {"host": None, "port": -1}, "application/json"
)
class DebuggerPluginLoader(base_plugin.TBLoader):
"""DebuggerPlugin factory factory.
This class determines which debugger plugin to load, based on custom
flags. It also checks for the `grpcio` PyPi dependency.
"""
def define_flags(self, parser):
"""Adds DebuggerPlugin CLI flags to parser."""
group = parser.add_argument_group("debugger plugin")
group.add_argument(
"--debugger_data_server_grpc_port",
metavar="PORT",
type=int,
default=-1,
help="""\
The port at which the non-interactive debugger data server should
return warning + make_table(html_arr)
def process_string_tensor_event(event):
"""Convert a TensorEvent into a JSON-compatible response."""
string_arr = tensor_util.make_ndarray(event.tensor_proto)
html = text_array_to_html(string_arr)
return {
"wall_time": event.wall_time,
"step": event.step,
"text": html,
}
class TextPlugin(base_plugin.TBPlugin):
"""Text Plugin for TensorBoard."""
plugin_name = metadata.PLUGIN_NAME
def __init__(self, context):
"""Instantiates TextPlugin via TensorBoard core.
Args:
context: A base_plugin.TBContext instance.
"""
self._multiplexer = context.multiplexer
def is_active(self):
"""Determines whether this plugin is active.
This plugin is only active if TensorBoard sampled any text summaries.