Coverage for python / lsst / pipe / base / tests / no_dimensions.py: 0%

1# This file is part of ctrl_mpexec. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://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 software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

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

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

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

18# (at your option) any later version. 

19# 

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

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

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

23# GNU General Public License for more details. 

24# 

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

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

27 

28from __future__ import annotations 

29 

30__all__ = ( 

31 "NoDimensionsTestConfig", 

32 "NoDimensionsTestConnections", 

33 "NoDimensionsTestTask", 

34) 

35 

36import copy 

37import dataclasses 

38from typing import cast 

39 

40from lsst.pex.config import Field 

41from lsst.pipe.base import ( 

42 PipelineTask, 

43 PipelineTaskConfig, 

44 PipelineTaskConnections, 

45 Struct, 

46 TaskMetadata, 

47 connectionTypes, 

48) 

49from lsst.utils.introspection import get_full_type_name 

50 

51 

52class NoDimensionsTestConnections(PipelineTaskConnections, dimensions=set()): 

53 """Connections class for `NoDimensionsTestTask`. 

54 

55 Parameters 

56 ---------- 

57 config : `PipelineTaskConfig` or `None`, optional 

58 Config to use for the connection. 

59 """ 

60 

61 input = connectionTypes.Input( 

62 name="input", doc="some dict-y input data for testing", storageClass="StructuredDataDict" 

63 ) 

64 output = connectionTypes.Output( 

65 name="output", doc="some dict-y output data for testing", storageClass="StructuredDataDict" 

66 ) 

67 

68 config: NoDimensionsTestConfig 

69 

70 def __init__(self, *, config: PipelineTaskConfig | None = None): 

71 if self.config.outputSC != "StructuredDataDict": 

72 self.output = dataclasses.replace(self.output, storageClass=self.config.outputSC) 

73 

74 

75class NoDimensionsTestConfig(PipelineTaskConfig, pipelineConnections=NoDimensionsTestConnections): 

76 """Configuration for `NoDimensionTestTask`.""" 

77 

78 key = Field[str](doc="String key for the dict entry the task sets.", default="one") 

79 value = Field[int](doc="Integer value for the dict entry the task sets.", default=1) 

80 outputSC = Field[str](doc="Output storage class requested", default="StructuredDataDict") 

81 

82 

83class NoDimensionsTestTask(PipelineTask): 

84 """A simple PipelineTask intended for tests that only need trivial 

85 relationships between tasks and datasets. 

86 

87 The quanta and input and output datasets of this task have no dimensions, 

88 so they use trivial, empty data IDs and require little data repository prep 

89 work to be used. 

90 """ 

91 

92 ConfigClass = NoDimensionsTestConfig 

93 _DefaultName = "noDimensionsTest" 

94 

95 # The completely flexible arguments to run aren't really valid inheritance; 

96 # the base class method exists just as a place to put a docstring, so we 

97 # tell mypy to ignore it. 

98 def run(self, input: TaskMetadata | dict[str, int]) -> Struct: 

99 """Run the task, adding the configured key-value pair to the input 

100 argument and returning it as the output. 

101 

102 Parameters 

103 ---------- 

104 input : `dict` 

105 Dictionary to update and return. 

106 

107 Returns 

108 ------- 

109 result : `lsst.pipe.base.Struct` 

110 Struct with a single ``output`` attribute. 

111 """ 

112 self.log.info("Run method given data of type: %s", get_full_type_name(input)) 

113 output = copy.copy(input) 

114 config = cast(NoDimensionsTestConfig, self.config) 

115 output[config.key] = config.value 

116 

117 # Can change the return type via configuration. 

118 if "TaskMetadata" in cast(NoDimensionsTestConfig, self.config).outputSC: 

119 output = TaskMetadata.from_dict(output) # type: ignore 

120 elif type(output) is TaskMetadata: 

121 # Want the output to be a dict 

122 output = output.to_dict() 

123 self.log.info("Run method returns data of type: %s", get_full_type_name(output)) 

124 return Struct(output=output)