Coverage for python / lsst / pipe / base / graph / graphSummary.py: 100%

1# This file is part of pipe_base. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21 

22import sys 

23from collections import Counter 

24from typing import Any 

25 

26import pydantic 

27 

28from .quantumNode import BuildId 

29 

30__all__ = ("QgraphSummary", "QgraphTaskSummary") 

31 

32 

33class QgraphTaskSummary(pydantic.BaseModel): 

34 """Quanta information summarized for single PipelineTask.""" 

35 

36 taskLabel: str | None = None 

37 """PipelineTask label.""" 

38 

39 numQuanta: int = 0 

40 """Number of Quanta for this PipelineTask in this QuantumGraph.""" 

41 

42 numInputs: dict[str, int] = pydantic.Field(default_factory=Counter) 

43 """Total number of inputs per dataset type name for this PipelineTask.""" 

44 

45 numOutputs: dict[str, int] = pydantic.Field(default_factory=Counter) 

46 """Total number of outputs per dataset type name for this PipelineTask.""" 

47 

48 # Work around the fact that Sphinx chokes on Pydantic docstring formatting, 

49 # when we inherit those docstrings in our public classes. 

50 if "sphinx" in sys.modules: # pragma: no cover 

51 

52 def copy(self, *args: Any, **kwargs: Any) -> Any: 

53 """See `pydantic.BaseModel.copy`.""" 

54 return super().copy(*args, **kwargs) 

55 

56 def model_dump(self, *args: Any, **kwargs: Any) -> Any: 

57 """See `pydantic.BaseModel.model_dump`.""" 

58 return super().model_dump(*args, **kwargs) 

59 

60 def model_dump_json(self, *args: Any, **kwargs: Any) -> Any: 

61 """See `pydantic.BaseModel.model_dump_json`.""" 

62 return super().model_dump_json(*args, **kwargs) 

63 

64 def model_copy(self, *args: Any, **kwargs: Any) -> Any: 

65 """See `pydantic.BaseModel.model_copy`.""" 

66 return super().model_copy(*args, **kwargs) 

67 

68 @classmethod 

69 def model_construct(cls, *args: Any, **kwargs: Any) -> Any: # type: ignore[override] 

70 """See `pydantic.BaseModel.model_construct`.""" 

71 return super().model_construct(*args, **kwargs) 

72 

73 @classmethod 

74 def model_json_schema(cls, *args: Any, **kwargs: Any) -> Any: 

75 """See `pydantic.BaseModel.model_json_schema`.""" 

76 return super().model_json_schema(*args, **kwargs) 

77 

78 @classmethod 

79 def model_validate(cls, *args: Any, **kwargs: Any) -> Any: 

80 """See `pydantic.BaseModel.model_validate`.""" 

81 return super().model_validate(*args, **kwargs) 

82 

83 @classmethod 

84 def model_validate_json(cls, *args: Any, **kwargs: Any) -> Any: 

85 """See `pydantic.BaseModel.model_validate_json`.""" 

86 return super().model_validate_json(*args, **kwargs) 

87 

88 @classmethod 

89 def model_validate_strings(cls, *args: Any, **kwargs: Any) -> Any: 

90 """See `pydantic.BaseModel.model_validate_strings`.""" 

91 return super().model_validate_strings(*args, **kwargs) 

92 

93 

94class QgraphSummary(pydantic.BaseModel): 

95 """Report for the QuantumGraph creation or reading.""" 

96 

97 graphID: BuildId | None = None 

98 """QuantumGraph ID.""" 

99 

100 cmdLine: str | None = None 

101 """Command line for creation of original QuantumGraph.""" 

102 

103 creationUTC: str | None = None 

104 """Time of creation.""" 

105 

106 inputCollection: list[str] | None = None 

107 """Input collection.""" 

108 

109 outputCollection: str | None = None 

110 """Output collection.""" 

111 

112 outputRun: str | None = None 

113 """Output run collection.""" 

114 

115 qgraphTaskSummaries: dict[str, QgraphTaskSummary] = pydantic.Field(default_factory=dict) 

116 """Quanta information summarized per PipelineTask.""" 

117 

118 # Work around the fact that Sphinx chokes on Pydantic docstring formatting, 

119 # when we inherit those docstrings in our public classes. 

120 if "sphinx" in sys.modules: # pragma: no cover 

121 

122 def copy(self, *args: Any, **kwargs: Any) -> Any: 

123 """See `pydantic.BaseModel.copy`.""" 

124 return super().copy(*args, **kwargs) 

125 

126 def model_dump(self, *args: Any, **kwargs: Any) -> Any: 

127 """See `pydantic.BaseModel.model_dump`.""" 

128 return super().model_dump(*args, **kwargs) 

129 

130 def model_dump_json(self, *args: Any, **kwargs: Any) -> Any: 

131 """See `pydantic.BaseModel.model_dump_json`.""" 

132 return super().model_dump_json(*args, **kwargs) 

133 

134 def model_copy(self, *args: Any, **kwargs: Any) -> Any: 

135 """See `pydantic.BaseModel.model_copy`.""" 

136 return super().model_copy(*args, **kwargs) 

137 

138 @classmethod 

139 def model_construct(cls, *args: Any, **kwargs: Any) -> Any: # type: ignore[override] 

140 """See `pydantic.BaseModel.model_construct`.""" 

141 return super().model_construct(*args, **kwargs) 

142 

143 @classmethod 

144 def model_json_schema(cls, *args: Any, **kwargs: Any) -> Any: 

145 """See `pydantic.BaseModel.model_json_schema`.""" 

146 return super().model_json_schema(*args, **kwargs) 

147 

148 @classmethod 

149 def model_validate(cls, *args: Any, **kwargs: Any) -> Any: 

150 """See `pydantic.BaseModel.model_validate`.""" 

151 return super().model_validate(*args, **kwargs) 

152 

153 @classmethod 

154 def model_validate_json(cls, *args: Any, **kwargs: Any) -> Any: 

155 """See `pydantic.BaseModel.model_validate_json`.""" 

156 return super().model_validate_json(*args, **kwargs) 

157 

158 @classmethod 

159 def model_validate_strings(cls, *args: Any, **kwargs: Any) -> Any: 

160 """See `pydantic.BaseModel.model_validate_strings`.""" 

161 return super().model_validate_strings(*args, **kwargs)