Coverage for python / lsst / pipe / base / quantum_graph / formatter.py: 86%

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 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 <http://www.gnu.org/licenses/>. 

27 

28from __future__ import annotations 

29 

30__all__ = ("ProvenanceFormatter",) 

31 

32import uuid 

33from typing import Any, ClassVar 

34 

35import pydantic 

36 

37from lsst.daf.butler import FormatterV2 

38from lsst.resources import ResourcePath 

39from lsst.utils.logging import getLogger 

40 

41from ..pipeline_graph import TaskImportMode 

42from ._provenance import ProvenanceQuantumGraphReader 

43 

44_LOG = getLogger(__file__) 

45 

46 

47class _ProvenanceFormatterParameters(pydantic.BaseModel): 

48 """A Pydantic model for validating and applying defaults to the 

49 read parameters of `ProvenanceFormatter`. 

50 """ 

51 

52 import_mode: TaskImportMode = TaskImportMode.DO_NOT_IMPORT 

53 quanta: list[uuid.UUID] | None = None 

54 datasets: list[uuid.UUID] | None = None 

55 read_init_quanta: bool = True 

56 

57 @pydantic.field_validator("quanta", mode="before") 

58 @classmethod 

59 def quanta_to_list(cls, v: Any) -> list[uuid.UUID]: 

60 return list(v) 

61 

62 @pydantic.field_validator("datasets", mode="before") 

63 @classmethod 

64 def datasets_to_list(cls, v: Any) -> list[uuid.UUID]: 

65 return list(v) 

66 

67 @property 

68 def nodes(self) -> list[uuid.UUID]: 

69 if self.quanta is not None: 69 ↛ 74line 69 didn't jump to line 74 because the condition on line 69 was always true

70 if self.datasets is not None: 70 ↛ 73line 70 didn't jump to line 73 because the condition on line 70 was always true

71 return self.quanta + self.datasets 

72 else: 

73 return self.quanta 

74 elif self.datasets is not None: 

75 return self.datasets 

76 raise ValueError("'datasets' and/or 'quanta' parameters are required for this component") 

77 

78 

79class ProvenanceFormatter(FormatterV2): 

80 """Butler interface for reading `ProvenanceQuantumGraph` objects.""" 

81 

82 default_extension: ClassVar[str] = ".qg" 

83 can_read_from_uri: ClassVar[bool] = True 

84 

85 def read_from_uri(self, uri: ResourcePath, component: str | None = None, expected_size: int = -1) -> Any: 

86 parameters = _ProvenanceFormatterParameters.model_validate(self.file_descriptor.parameters or {}) 

87 with ProvenanceQuantumGraphReader.open(uri, import_mode=parameters.import_mode) as reader: 

88 match component: 

89 case None: 

90 if parameters.read_init_quanta: 90 ↛ 92line 90 didn't jump to line 92 because the condition on line 90 was always true

91 reader.read_init_quanta() 

92 reader.read_quanta(parameters.quanta) 

93 reader.read_datasets(parameters.datasets) 

94 return reader.graph 

95 case "metadata": 

96 return reader.fetch_metadata(parameters.nodes) 

97 case "logs": 

98 return reader.fetch_logs(parameters.nodes) 

99 case "packages": 

100 return reader.fetch_packages() 

101 raise AssertionError(f"Unexpected component {component!r}.")