Coverage for python / lsst / daf / butler / script / collectionChain.py: 10%
52 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-26 08:49 +0000
1# This file is part of daf_butler.
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/>.
28from __future__ import annotations
30from collections.abc import Iterable
32from .._butler import Butler
33from .._collection_type import CollectionType
34from ..registry import MissingCollectionError
37def collectionChain(
38 repo: str, mode: str, parent: str, children: Iterable[str], doc: str | None, flatten: bool
39) -> tuple[str, ...]:
40 """Get the collections whose names match an expression.
42 Parameters
43 ----------
44 repo : `str`
45 URI to the location of the repo or URI to a config file describing the
46 repo and its location.
47 mode : `str`
48 Update mode for this chain. Options are:
49 'redefine': Create or modify ``parent`` to be defined by the supplied
50 ``children``.
51 'remove': Modify existing chain to remove ``children`` from it.
52 'prepend': Add the given ``children`` to the beginning of the chain.
53 'extend': Modify existing chain to add ``children`` to the end of it.
54 'pop': Pop a numbered element off the chain. Defaults to popping
55 the first element (0). ``children`` must be integers if given.
56 Both 'prepend' and 'extend' are the same as 'redefine' if the chain
57 does not exist.
58 parent : `str`
59 Name of the chained collection to update. Will be created if it
60 does not exist already.
61 children : `~collections.abc.Iterable` of `str`
62 Names of the children to be included in the chain.
63 doc : `str`
64 If the chained collection is being created, the documentation string
65 that will be associated with it.
66 flatten : `bool`
67 If `True`, recursively flatten out any nested
68 `~CollectionType.CHAINED` collections in ``children`` first.
70 Returns
71 -------
72 chain : `tuple` of `str`
73 The collections in the chain following this command.
74 """
75 with Butler.from_config(repo, writeable=True, without_datastore=True) as butler:
76 # Every mode needs children except pop.
77 if not children and mode != "pop":
78 raise RuntimeError(f"Must provide children when defining a collection chain in mode {mode}.")
80 try:
81 butler.collections.get_info(parent)
82 except MissingCollectionError:
83 # Create it -- but only if mode can work with empty chain.
84 if mode in ("redefine", "extend", "prepend"):
85 if not doc:
86 doc = None
87 butler.collections.register(parent, CollectionType.CHAINED, doc)
88 else:
89 raise RuntimeError(
90 f"Mode '{mode}' requires that the collection exists "
91 f"but collection '{parent}' is not known to this registry"
92 ) from None
94 if flatten:
95 if mode not in ("redefine", "prepend", "extend"):
96 raise RuntimeError(f"'flatten' flag is not allowed for {mode}")
97 children = butler.collections.query(children, flatten_chains=True)
99 _modify_collection_chain(butler, mode, parent, children)
101 return butler.collections.get_info(parent).children
104def _modify_collection_chain(butler: Butler, mode: str, parent: str, children: Iterable[str]) -> None:
105 if mode == "prepend":
106 butler.collections.prepend_chain(parent, children)
107 elif mode == "redefine":
108 butler.collections.redefine_chain(parent, children)
109 elif mode == "remove":
110 butler.collections.remove_from_chain(parent, children)
111 elif mode == "pop":
112 children_to_pop = _find_children_to_pop(butler, parent, children)
113 butler.collections.remove_from_chain(parent, children_to_pop)
114 elif mode == "extend":
115 butler.collections.extend_chain(parent, children)
116 else:
117 raise ValueError(f"Unrecognized update mode: '{mode}'")
120def _find_children_to_pop(butler: Butler, parent: str, children: Iterable[str]) -> list[str]:
121 """Find the names of the children in the parent collection corresponding to
122 the given indexes.
123 """
124 children = list(children)
125 collection_info = butler.collections.get_info(parent)
126 current = collection_info.children
127 n_current = len(current)
128 if children:
130 def convert_index(i: int) -> int:
131 """Convert negative index to positive."""
132 if i >= 0:
133 return i
134 return n_current + i
136 # For this mode the children should be integers.
137 # Convert negative integers to positive ones to allow
138 # sorting.
139 indices = [convert_index(int(child)) for child in children]
140 else:
141 # Nothing specified, pop from the front of the chain.
142 indices = [0]
144 for index in indices:
145 if index >= n_current:
146 raise IndexError(
147 f"Index {index} is out of range. Parent collection {parent} has {n_current} children."
148 )
150 return [current[i] for i in indices]