22 from __future__
import annotations
24 __all__ = (
"Translator",
"TranslatorFactory",
"KeyHandler",
"CopyKeyHandler",
"ConstantKeyHandler",
25 "BandToPhysicalFilterKeyHandler",
"PhysicalFilterToBandKeyHandler")
28 from typing
import Optional, Any, Dict, Tuple, FrozenSet, Iterable, List
29 from abc
import ABCMeta, abstractmethod
36 """Base class for Translator helpers that each handle just one Gen3 Data
42 Name of the Gen3 dimension (data ID key) populated by
43 this handler (e.g. "visit" or "band").
48 __slots__ = (
"dimension",)
51 return f
"{type(self).__name__}({self.dimension}, ...)"
54 skyMap: Optional[BaseSkyMap], skyMapName: Optional[str],
55 datasetTypeName: str):
56 """Update a Gen3 data ID dict with a single key-value pair from a Gen2
59 This method is implemented by the base class and is not expected to
60 be re-implemented by subclasses.
65 Gen2 data ID from which to draw key-value pairs from.
67 Gen3 data ID to update in-place.
68 skyMap: `BaseSkyMap`, optional
69 SkyMap that defines the tracts and patches used in the Gen2 data
72 Name of the Gen3 skymap dimension that defines the tracts and
73 patches used in the Gen3 data ID.
74 datasetTypeName: `str`
75 Name of the dataset type.
77 gen3id[self.
dimension] = self.
extract(gen2id, skyMap=skyMap, skyMapName=skyMapName,
78 datasetTypeName=datasetTypeName)
81 def extract(self, gen2id: dict, skyMap: Optional[BaseSkyMap], skyMapName: Optional[str],
82 datasetTypeName: str) -> Any:
83 """Extract a Gen3 data ID value from a Gen2 data ID.
88 Gen2 data ID from which to draw key-value pairs from.
89 skyMap: `BaseSkyMap`, optional
90 SkyMap that defines the tracts and patches used in the Gen2 data
93 Name of the Gen3 skymap dimension that defines the tracts and
94 patches used in the Gen3 data ID.
95 datasetTypeName: `str`
96 Name of the dataset type.
98 raise NotImplementedError()
102 """A KeyHandler that adds a constant key-value pair to the Gen3 data ID.
107 Name of the Gen3 dimension (data ID key) populated by
108 this handler (e.g. "visit" or "band").
116 __slots__ = (
"value",)
118 def extract(self, gen2id: dict, skyMap: Optional[BaseSkyMap], skyMapName: Optional[str],
119 datasetTypeName: str) -> Any:
125 """A KeyHandler that simply copies a value from a Gen3 data ID.
130 Name of the Gen3 dimension produced by this handler.
131 dtype : `type`, optional
132 If not `None`, the type that values for this key must be an
135 def __init__(self, dimension: str, gen2key: Optional[str] =
None,
136 dtype: Optional[type] =
None):
138 self.
gen2key = gen2key
if gen2key
is not None else dimension
141 __slots__ = (
"gen2key",
"dtype")
144 return f
"{type(self).__name__}({self.gen2key}, {self.dtype})"
146 def extract(self, gen2id: dict, skyMap: Optional[BaseSkyMap], skyMapName: Optional[str],
147 datasetTypeName: str) -> Any:
150 if self.
dtype is not None:
153 except ValueError
as err:
155 f
"'{r}' is not a valid value for {self.dimension}; "
156 f
"expected {self.dtype.__name__}, got {type(r).__name__}."
162 """A KeyHandler for skymap patches.
169 def extract(self, gen2id: dict, skyMap: Optional[BaseSkyMap], skyMapName: Optional[str],
170 datasetTypeName: str) -> Any:
172 tract = gen2id[
"tract"]
173 tractInfo = skyMap[tract]
174 x, y = gen2id[
"patch"].split(
",")
175 patchInfo = tractInfo[int(x), int(y)]
176 return tractInfo.getSequentialPatchIndex(patchInfo)
180 """A KeyHandler for skymaps."""
186 def extract(self, gen2id: dict, skyMap: Optional[BaseSkyMap], skyMapName: Optional[str],
187 datasetTypeName: str) -> Any:
193 """KeyHandler for gen2 ``filter`` keys that match ``physical_filter``
194 keys in gen3 but should be mapped to ``band``.
196 Note that multiple physical filter can potentially map to one abstract
197 filter, so be careful to only use this translator on obs packages where
198 there is a one-to-one mapping.
201 __slots__ = (
"_map",)
205 self.
_map = {d.physical_filter: d.band
for d
in filterDefinitions
206 if d.physical_filter
is not None}
209 physical = gen2id[
"filter"]
210 return self.
_map.get(physical, physical)
214 """KeyHandler for gen2 ``filter`` keys that match ``band``
215 keys in gen3 but should be mapped to ``physical_filter``.
217 Note that one abstract filter can potentially map to multiple physical
218 filters, so be careful to only use this translator on obs packages where
219 there is a one-to-one mapping.
222 __slots__ = (
"_map",)
226 self.
_map = {d.band: d.physical_filter
for d
in filterDefinitions
227 if d.band
is not None}
230 abstract = gen2id[
"filter"]
231 return self.
_map.get(abstract, abstract)
235 """A class that manages a system of rules for translating Gen2 data IDs
236 to Gen3 data IDs, and uses these to construct translators for particular
241 log : `lsst.log.Log`, optional
242 A logger for diagnostic messages.
251 List[Tuple[FrozenSet[str], KeyHandler, bool]]
260 log = Log.getLogger(
"obs.base.gen2to3.TranslatorFactory")
265 for instrumentName, nested
in self._rules.
items():
266 if instrumentName
is None:
267 instrumentName =
"[any instrument]"
268 for datasetTypeName, rules
in nested.items():
269 if datasetTypeName
is None:
270 datasetTypeName =
"[any dataset type]"
271 lines.append(f
"{instrumentName} + {datasetTypeName}:")
272 for gen2keys, handler, consume
in rules:
273 consumed =
" (consumed)" if consume
else ""
274 lines.append(f
" {gen2keys}{consumed}: {handler}")
275 return "\n".join(lines)
277 def addRule(self, handler: KeyHandler, instrument: Optional[str] =
None,
278 datasetTypeName: Optional[str] =
None, gen2keys: Iterable[str] = (),
279 consume: bool =
True):
280 """Add a KeyHandler and an associated matching rule.
284 handler : `KeyHandler`
285 A KeyHandler instance to add to a Translator when this rule
288 Gen3 instrument name the Gen2 repository must be associated with
289 for this rule to match, or None to match any instrument.
290 datasetTypeName : `str`
291 Name of the DatasetType this rule matches, or None to match any
294 Sequence of Gen2 data ID keys that must all be present for this
296 consume : `bool` or `tuple`
297 If True (default), remove all entries in gen2keys from the set of
298 keys being matched to in order to prevent less-specific handlers
300 May also be a `tuple` listing only the keys to consume.
305 consume = frozenset(gen2keys)
307 consume = frozenset(consume)
309 consume = frozenset()
313 rulesForInstrument = self._rules.setdefault(instrument, {
None: []})
314 rulesForInstrumentAndDatasetType = rulesForInstrument.setdefault(datasetTypeName, [])
315 rulesForInstrumentAndDatasetType.append((frozenset(gen2keys), handler, consume))
317 def _addDefaultRules(self):
318 """Add translator rules that should always be present, and don't depend
319 at all on the instrument whose datasets are being converted.
321 This is called by `TranslatorFactory` construction.
327 for coaddName
in (
"deep",
"goodSeeing",
"psfMatched",
"dcr"):
342 gen2keys=(
"filter",
"tract"),
359 calibFilterType: str =
"physical_filter",
360 detectorKey: str =
"ccd",
361 exposureKey: str =
"visit"):
362 """Add translation rules that depend on some properties of the
363 instrument but are otherwise generic.
368 The short (dimension) name of the instrument that conversion is
370 calibFilterType : `str`, optional
371 One of ``physical_filter`` or ``band``, indicating which
372 of those the gen2 calibRegistry uses as the ``filter`` key.
373 detectorKey : `str`, optional
374 The gen2 key used to identify what in gen3 is `detector`.
375 exposureKey : `str`, optional
376 The gen2 key used to identify what in gen3 is `exposure`.
382 instrument=instrumentName, gen2keys=(exposureKey,), consume=
False)
384 instrument=instrumentName, gen2keys=(detectorKey,), consume=
False)
386 instrument=instrumentName, gen2keys=(
"calibDate",), consume=
False)
391 instrument=instrumentName, datasetTypeName=
"raw", gen2keys=(exposureKey,),
392 consume=(exposureKey,
"filter"))
396 consume=(
"visit",
"filter"))
400 instrument=instrumentName,
401 gen2keys=(detectorKey,))
406 instrument=instrumentName, datasetTypeName=
"transmission_optics")
408 instrument=instrumentName, datasetTypeName=
"transmission_atmosphere")
410 instrument=instrumentName, datasetTypeName=
"transmission_filter")
412 instrument=instrumentName, datasetTypeName=
"transmission_filter")
415 for calibType
in (
'flat',
'sky',
'fringe'):
417 instrument=instrumentName, datasetTypeName=calibType)
419 def makeMatching(self, datasetTypeName: str, gen2keys: Dict[str, type], instrument: Optional[str] =
None,
420 skyMap: Optional[BaseSkyMap] =
None, skyMapName: Optional[str] =
None):
421 """Construct a Translator appropriate for instances of the given
426 datasetTypeName : `str`
427 Name of the dataset type.
429 Keys of a Gen2 data ID for this dataset.
430 instrument: `str`, optional
431 Name of the Gen3 instrument dimension for translated data IDs.
432 skyMap: `~lsst.skymap.BaseSkyMap`, optional
433 The skymap instance that defines any tract/patch data IDs.
434 `~lsst.skymap.BaseSkyMap` instances.
435 skyMapName : `str`, optional
436 Gen3 SkyMap Dimension name to be associated with any tract or patch
441 translator : `Translator`
442 A translator whose translate() method can be used to transform Gen2
443 data IDs to Gen3 dataIds.
445 if instrument
is not None:
446 rulesForInstrument = self._rules.get(instrument, {
None: []})
448 rulesForInstrument = {
None: []}
449 rulesForAnyInstrument = self._rules[
None]
450 candidateRules = itertools.chain(
451 rulesForInstrument.get(datasetTypeName, []),
452 rulesForInstrument[
None],
453 rulesForAnyInstrument.get(datasetTypeName, []),
454 rulesForAnyInstrument[
None],
457 targetKeys =
set(gen2keys)
458 self.
log.
debug(
"Constructing data ID translator for %s with Gen2 keys %s...",
459 datasetTypeName, gen2keys)
460 for ruleKeys, ruleHandlers, consume
in candidateRules:
461 if ruleKeys.issubset(targetKeys):
462 matchedHandlers.append(ruleHandlers)
463 targetKeys -= consume
464 self.
log.
debug(
"...matched %d handlers: %s, with %s unmatched.",
465 len(matchedHandlers), matchedHandlers, targetKeys)
466 return Translator(matchedHandlers, skyMap=skyMap, skyMapName=skyMapName,
467 datasetTypeName=datasetTypeName, log=self.
log)
471 """Callable object that translates Gen2 Data IDs to Gen3 Data IDs for a
472 particular DatasetType.
474 Translators should usually be constructed via
475 `TranslatorFactory.makeMatching`.
480 A list of KeyHandlers this Translator should use.
481 skyMap : `BaseSkyMap`, optional
482 SkyMap instance used to define any tract or patch Dimensions.
484 Gen3 SkyMap Dimension name to be associated with any tract or patch
486 datasetTypeName : `str`
487 Name of the dataset type whose data IDs this translator handles.
489 def __init__(self, handlers: List[KeyHandler], skyMap: Optional[BaseSkyMap], skyMapName: Optional[str],
490 datasetTypeName: str, log: Log):
497 __slots__ = (
"handlers",
"skyMap",
"skyMapName",
"datasetTypeName",
"log")
500 hstr =
",".join(str(h)
for h
in self.
handlers)
501 return f
"{type(self).__name__}(dtype={self.datasetTypeName}, handlers=[{hstr}])"
503 def __call__(self, gen2id: Dict[str, Any], *, partial: bool =
False) -> Tuple[dict, Optional[str]]:
504 """Return a Gen3 data ID that corresponds to the given Gen2 data ID.
507 calibDate = gen2id.get(
"calibDate",
None)
510 handler.translate(gen2id, gen3id, skyMap=self.
skyMap, skyMapName=self.
skyMapName,
514 self.
log.
debug(
"Failed to translate %s from %s (this may not be an error).",
515 handler.dimension, gen2id)
519 return gen3id, calibDate
523 """The names of the dimensions populated by this Translator
526 return frozenset(h.dimension
for h
in self.
handlers)