Coverage for python / lsst / pipe / base / script / zip_from_graph.py: 50%

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 

28__all__ = ["zip_from_graph"] 

29 

30import logging 

31 

32from lsst.daf.butler import QuantumBackedButler 

33from lsst.pipe.base import QuantumGraph 

34from lsst.resources import ResourcePath 

35 

36from .utils import filter_by_dataset_type_glob 

37 

38_LOG = logging.getLogger(__name__) 

39 

40 

41def zip_from_graph( 

42 graph: str, 

43 repo: str, 

44 dest: str, 

45 dataset_type: tuple[str, ...], 

46) -> ResourcePath: 

47 """Create Zip export file from graph outputs. 

48 

49 Parameters 

50 ---------- 

51 graph : `str` 

52 URI string of the quantum graph. 

53 repo : `str` 

54 URI to a butler configuration used to define the datastore associated 

55 with the graph. 

56 dest : `str` 

57 Path to the destination directory for the Zip file. 

58 dataset_type : `tuple` of `str` 

59 Dataset type names. An empty tuple implies all dataset types. 

60 Can include globs. 

61 

62 Returns 

63 ------- 

64 zip_path : `lsst.resources.ResourcePath` 

65 Path to the Zip file. 

66 """ 

67 # Read whole graph into memory 

68 qgraph = QuantumGraph.loadUri(graph) 

69 

70 output_refs, _ = qgraph.get_refs(include_outputs=True, include_init_outputs=True, conform_outputs=True) 

71 

72 # Get data repository dataset type definitions from the QuantumGraph. 

73 dataset_types = {dstype.name: dstype for dstype in qgraph.registryDatasetTypes()} 

74 

75 # Filter the refs based on requested dataset types. 

76 filtered_refs = filter_by_dataset_type_glob(output_refs, dataset_type) 

77 

78 # Make QBB, its config is the same as output Butler. 

79 with QuantumBackedButler.from_predicted( 

80 config=repo, 

81 predicted_inputs=[ref.id for ref in output_refs], 

82 predicted_outputs=[], 

83 dimensions=qgraph.universe, 

84 datastore_records={}, 

85 dataset_types=dataset_types, 

86 ) as qbb: 

87 _LOG.info("Retrieving artifacts for %d datasets and storing in Zip file.", len(filtered_refs)) 

88 zip = qbb.retrieve_artifacts_zip(filtered_refs, dest) 

89 return zip