Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/tensorboard/summary/_writer.py: 36%

1# Copyright 2021 The TensorFlow Authors. All Rights Reserved. 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================== 

15"""Implementation for tensorboard.summary.Writer and related symbols. 

16 

17This provides a TensorBoard-native summary writing API that only depends 

18on numpy and not any particular ML framework. 

19""" 

20 

21import time 

22 

23import numpy as np 

24 

25from tensorboard.plugins.scalar import metadata as scalars_metadata 

26from tensorboard.summary import _output 

27 

28 

29class Writer: 

30 """Writes summary data for visualization in TensorBoard. 

31 

32 This class is not thread-safe. 

33 

34 TODO(#4581): This API should be considered EXPERIMENTAL and subject to 

35 backwards-incompatible changes without notice. 

36 """ 

37 

38 def __init__(self, output): 

39 """Constructs a Writer. 

40 

41 Args: 

42 output: `tensorboard.summary.Output` object, or a string which will be 

43 interpreted as shorthand for an `Output` of the appropriate type. The 

44 only currently supported type is `DirectoryOutput`, where the string 

45 value given here will be used as the directory path. 

46 """ 

47 if isinstance(output, _output.Output): 

48 self._output = output 

49 elif isinstance(output, str): 

50 self._output = _output.DirectoryOutput(output) 

51 else: 

52 raise TypeError("Unsupported output object %r" % output) 

53 self._closed = False 

54 

55 def _check_not_closed(self): 

56 if self._closed: 

57 raise RuntimeError("Writer is already closed") 

58 

59 def flush(self): 

60 """Flushes any buffered data.""" 

61 self._check_not_closed() 

62 self._output.flush() 

63 

64 def close(self): 

65 """Closes the writer and prevents further use.""" 

66 self._check_not_closed() 

67 self._output.close() 

68 self._closed = True 

69 

70 def add_scalar(self, tag, data, step, *, wall_time=None, description=None): 

71 """Adds a scalar summary. 

72 

73 Args: 

74 tag: string tag used to uniquely identify this time series. 

75 data: numeric scalar value for this data point. Accepts any value that 

76 can be converted to a `np.float32` scalar. 

77 step: integer step value for this data point. Accepts any value that 

78 can be converted to a `np.int64` scalar. 

79 wall_time: optional `float` seconds since the Unix epoch, representing 

80 the real-world timestamp for this data point. Defaults to None in 

81 which case the current time will be used. 

82 description: optional string description for this entire time series. 

83 This should be constant for a given tag; only the first value 

84 encountered will be used. 

85 """ 

86 self._check_not_closed() 

87 validated_data = _validate_scalar_shape(np.float32(data), "data") 

88 validated_step = _validate_scalar_shape(np.int64(step), "step") 

89 wall_time = wall_time if wall_time is not None else time.time() 

90 self._output.emit_scalar( 

91 plugin_name=scalars_metadata.PLUGIN_NAME, 

92 tag=tag, 

93 data=validated_data, 

94 step=validated_step, 

95 wall_time=wall_time, 

96 description=description, 

97 ) 

98 

99 

100def _validate_scalar_shape(ndarray, name): 

101 if ndarray.ndim != 0: 

102 raise ValueError( 

103 "Expected scalar value for %r but got %r" % (name, ndarray) 

104 ) 

105 return ndarray