Coverage for python / lsst / ctrl / mpexec / cli / script / pre_exec_init_qbb.py: 26%

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# (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 

30from lsst.pipe.base import BuildId, QuantumGraph 

31from lsst.pipe.base.pipeline_graph import TaskImportMode 

32from lsst.pipe.base.quantum_graph import PredictedQuantumGraph 

33from lsst.resources import ResourcePath, ResourcePathExpression 

34from lsst.utils.logging import getLogger 

35 

36from ..butler_factory import ButlerFactory 

37 

38_LOG = getLogger(__name__) 

39 

40 

41def pre_exec_init_qbb( 

42 butler_config: str, 

43 qgraph: ResourcePathExpression, 

44 qgraph_id: str | None, 

45 config_search_path: list[str] | None, 

46 **kwargs: object, 

47) -> None: 

48 """Implement the command line interface ``pipetask pre-exec-init-qbb`` 

49 subcommand. 

50 

51 Should only be called by command line tools and unit test code 

52 that tests this function. 

53 

54 Parameters 

55 ---------- 

56 butler_config : `str` 

57 The path location of the gen3 butler/registry config file. 

58 qgraph : `str` 

59 URI location for a serialized quantum graph definition. 

60 qgraph_id : `str` or `None` 

61 Quantum graph identifier, if specified must match the identifier of the 

62 graph loaded from a file. Ignored if graph is not loaded from a file. 

63 config_search_path : `list` [`str`] 

64 Additional search paths for butler configuration. 

65 **kwargs : `object` 

66 Ignored; click commands may accept options for more than one script 

67 function and pass all the option kwargs to each of the script functions 

68 which ignore these unused kwargs. 

69 """ 

70 qgraph = ResourcePath(qgraph) 

71 match qgraph.getExtension(): 

72 case ".qgraph": 

73 _LOG.verbose("Reading full quantum graph from %s.", qgraph) 

74 qg = PredictedQuantumGraph.from_old_quantum_graph( 

75 QuantumGraph.loadUri( 

76 qgraph, 

77 graphID=BuildId(qgraph_id) if qgraph_id is not None else None, 

78 ) 

79 ) 

80 case ".qg": 

81 _LOG.verbose("Reading init quanta from quantum graph from %s.", qgraph) 

82 if qgraph_id is not None: 

83 _LOG.warning("--qgraph-id is ignored when loading new '.qg' files.") 

84 with PredictedQuantumGraph.open( 

85 qgraph, import_mode=TaskImportMode.ASSUME_CONSISTENT_EDGES 

86 ) as reader: 

87 reader.read_init_quanta() 

88 qg = reader.finish() 

89 case ext: 

90 raise ValueError(f"Unrecognized extension for quantum graph: {ext!r}") 

91 

92 # Ensure that QBB uses shared datastore cache for writes. 

93 ButlerFactory.define_datastore_cache() 

94 

95 # Make QBB. 

96 _LOG.verbose("Initializing quantum-backed butler.") 

97 with qg.make_init_qbb(butler_config, config_search_paths=config_search_path) as butler: 

98 # Save all InitOutputs, configs, etc. 

99 _LOG.verbose("Instantiating tasks and saving init-outputs.") 

100 qg.init_output_run(butler)