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

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"""Generalized output options for writing tensor-formatted summary data.""" 

16 

17from tensorboard.compat.proto import event_pb2 

18from tensorboard.compat.proto import summary_pb2 

19from tensorboard.summary.writer import event_file_writer 

20from tensorboard.util import tensor_util 

21 

22import abc 

23 

24 

25class Output(abc.ABC): 

26 """Interface for emitting tensor-formatted summary data. 

27 

28 Implementations of this interface can be passed to Writer to customize 

29 how summary data is actually persisted (e.g. to disk, to memory, over 

30 the network, etc.). 

31 

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

33 backwards-incompatible changes without notice. 

34 """ 

35 

36 @abc.abstractmethod 

37 def emit_scalar( 

38 self, 

39 *, 

40 plugin_name, 

41 tag, 

42 data, 

43 step, 

44 wall_time, 

45 tag_metadata=None, 

46 description=None, 

47 ): 

48 """Emits one scalar data point to this Output. 

49 

50 Args: 

51 plugin_name: string name to uniquely identify the type of time series 

52 (historically associated with a TensorBoard plugin). 

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

54 data: `np.float32` scalar value for this data point. 

55 step: `np.int64` scalar step value for this data point. 

56 wall_time: `float` seconds since the Unix epoch, representing the 

57 real-world timestamp for this data point. 

58 tag_metadata: optional bytes containing metadata for this entire time 

59 series. This should be constant for a given tag; only the first 

60 value encountered will be used. 

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

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

63 encountered will be used. 

64 """ 

65 pass 

66 

67 @abc.abstractmethod 

68 def flush(self): 

69 """Flushes any data that has been buffered.""" 

70 pass 

71 

72 @abc.abstractmethod 

73 def close(self): 

74 """Closes the Output and also flushes any buffered data.""" 

75 pass 

76 

77 

78class DirectoryOutput(Output): 

79 """Outputs summary data by writing event files to a log directory. 

80 

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

82 backwards-incompatible changes without notice. 

83 """ 

84 

85 def __init__(self, path): 

86 """Creates a `DirectoryOutput` for the given path.""" 

87 self._ev_writer = event_file_writer.EventFileWriter(path) 

88 

89 def emit_scalar( 

90 self, 

91 *, 

92 plugin_name, 

93 tag, 

94 data, 

95 step, 

96 wall_time, 

97 tag_metadata=None, 

98 description=None, 

99 ): 

100 """See `Output`.""" 

101 # TODO(#4581): cache summary metadata to emit only once. 

102 summary_metadata = summary_pb2.SummaryMetadata( 

103 plugin_data=summary_pb2.SummaryMetadata.PluginData( 

104 plugin_name=plugin_name, content=tag_metadata 

105 ), 

106 summary_description=description, 

107 data_class=summary_pb2.DataClass.DATA_CLASS_SCALAR, 

108 ) 

109 tensor_proto = tensor_util.make_tensor_proto(data) 

110 event = event_pb2.Event(wall_time=wall_time, step=step) 

111 event.summary.value.add( 

112 tag=tag, tensor=tensor_proto, metadata=summary_metadata 

113 ) 

114 self._ev_writer.add_event(event) 

115 

116 def flush(self): 

117 """See `Output`.""" 

118 self._ev_writer.flush() 

119 

120 def close(self): 

121 """See `Output`.""" 

122 # No need to call flush first since EventFileWriter already 

123 # will do this for us when we call close(). 

124 self._ev_writer.close()