Coverage for python / lsst / daf / butler / _formatter.py: 23%
531 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-05 01:18 -0800
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
30__all__ = (
31 "FileIntegrityError",
32 "Formatter",
33 "FormatterFactory",
34 "FormatterNotImplementedError",
35 "FormatterParameter",
36 "FormatterV1inV2",
37 "FormatterV2",
38)
40import contextlib
41import copy
42import logging
43import os
44import zipfile
45from abc import ABCMeta, abstractmethod
46from collections.abc import Callable, Iterator, Mapping, Set
47from typing import TYPE_CHECKING, Any, BinaryIO, ClassVar, TypeAlias, final
49from lsst.resources import ResourceHandleProtocol, ResourcePath
50from lsst.utils.introspection import get_full_type_name
51from lsst.utils.timer import time_this
53from ._config import Config
54from ._config_support import LookupKey, processLookupConfigs
55from ._file_descriptor import FileDescriptor
56from ._location import Location
57from ._rubin.temporary_for_ingest import TemporaryForIngest
58from .dimensions import DataCoordinate, DimensionUniverse
59from .mapping_factory import MappingFactory
61log = logging.getLogger(__name__)
63if TYPE_CHECKING:
64 from ._dataset_provenance import DatasetProvenance
65 from ._dataset_ref import DatasetRef
66 from ._dataset_type import DatasetType
67 from ._storage_class import StorageClass
68 from .datastore.cache_manager import AbstractDatastoreCacheManager
70 # Define a new special type for functions that take "entity"
71 Entity: TypeAlias = DatasetType | DatasetRef | StorageClass | str
74class FileIntegrityError(RuntimeError):
75 """The file metadata is inconsistent with the metadata supplied by
76 the datastore.
77 """
80class FormatterNotImplementedError(NotImplementedError):
81 """Formatter does not implement the specific read or write method
82 that is being requested.
83 """
86class FormatterV2:
87 """Interface for reading and writing datasets using URIs.
89 The formatters are associated with a particular `StorageClass`.
91 Parameters
92 ----------
93 file_descriptor : `FileDescriptor`, optional
94 Identifies the file to read or write, and the associated storage
95 classes and parameter information.
96 ref : `DatasetRef`
97 The dataset associated with this formatter. Should not be a component
98 dataset ref.
99 write_parameters : `dict`, optional
100 Parameters to control how the dataset is serialized.
101 write_recipes : `dict`, optional
102 Detailed write recipes indexed by recipe name.
104 **kwargs
105 Additional arguments that will be ignored but allow for
106 `Formatter` V1 parameters to be given.
108 Notes
109 -----
110 A `FormatterV2` author should not override the default `read` or `write`
111 method. Instead for read the formatter author should implement one or all
112 of `read_from_stream`, `read_from_uri`, or `read_from_local_file`. The
113 method `read_from_uri` will always be attempted first and could be more
114 efficient (since it allows the possibility for a subset of the data file to
115 be accessed remotely when parameters or components are specified) but it
116 will not update the local cache. If the entire contents of the remote file
117 are being accessed (no component or parameters defined) and the dataset
118 would be cached, `read_from_uri` will be called with a local file. If the
119 file is remote and the parameters that have been included are known to be
120 more efficiently handled with a local file, the `read_from_uri` method can
121 return `NotImplemented` to indicate that a local file should be given
122 instead.
124 Similarly for writes, the `write` method can not be subclassed. Instead
125 the formatter author should implement `to_bytes` or `write_local_file`.
126 For local URIs the system will always call `write_local_file` first (which
127 by default will call `to_bytes`) to ensure atomic writes are implemented.
128 For remote URIs with local caching disabled, `to_bytes` will be called
129 first and the remote updated directly. If the dataset should be cached
130 it will always be written locally first.
131 """
133 unsupported_parameters: ClassVar[Set[str] | None] = frozenset()
134 """Set of read parameters not understood by this `Formatter`. An empty set
135 means all parameters are supported. `None` indicates that no parameters
136 are supported. These parameters should match those defined in the storage
137 class definition. (`frozenset`).
138 """
140 supported_write_parameters: ClassVar[Set[str] | None] = None
141 """Parameters understood by this formatter that can be used to control
142 how a dataset is serialized. `None` indicates that no parameters are
143 supported."""
145 default_extension: ClassVar[str | None] = None
146 """Default extension to use when writing a file.
148 Can be `None` if the extension is determined dynamically. Use the
149 `get_write_extension` method to get the actual extension to use.
150 """
152 supported_extensions: ClassVar[Set[str]] = frozenset()
153 """Set of all extensions supported by this formatter.
155 Any extension assigned to the ``default_extension`` property will be
156 automatically included in the list of supported extensions.
157 """
159 can_read_from_uri: ClassVar[bool] = False
160 """Declare whether `read_from_uri` is available to this formatter."""
162 can_read_from_stream: ClassVar[bool] = False
163 """Declare whether `read_from_stream` is available to this formatter."""
165 can_read_from_local_file: ClassVar[bool] = False
166 """Declare whether `read_from_file` is available to this formatter."""
168 def __init__(
169 self,
170 file_descriptor: FileDescriptor,
171 *,
172 ref: DatasetRef,
173 write_parameters: Mapping[str, Any] | None = None,
174 write_recipes: Mapping[str, Any] | None = None,
175 # Compatibility parameters. Unused in v2.
176 **kwargs: Any,
177 ):
178 if not isinstance(file_descriptor, FileDescriptor):
179 raise TypeError("File descriptor must be a FileDescriptor")
181 self._file_descriptor = file_descriptor
183 if ref.isComponent():
184 # It is a component ref for disassembled composites.
185 ref = ref.makeCompositeRef()
186 self._dataset_ref = ref
188 # Check that the write parameters are allowed
189 if write_parameters:
190 if self.supported_write_parameters is None:
191 raise ValueError(
192 f"This formatter does not accept any write parameters. Got: {', '.join(write_parameters)}"
193 )
194 else:
195 given = set(write_parameters)
196 unknown = given - self.supported_write_parameters
197 if unknown:
198 s = "s" if len(unknown) != 1 else ""
199 unknownStr = ", ".join(f"'{u}'" for u in unknown)
200 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}")
202 self._write_parameters = write_parameters
203 self._write_recipes = self.validate_write_recipes(write_recipes)
205 def __str__(self) -> str:
206 return f"{self.name()}@{self.file_descriptor.location.uri}"
208 def __repr__(self) -> str:
209 return f"{self.name()}({self.file_descriptor!r})"
211 @property
212 def file_descriptor(self) -> FileDescriptor:
213 """File descriptor associated with this formatter
214 (`FileDescriptor`).
215 """
216 return self._file_descriptor
218 @property
219 def data_id(self) -> DataCoordinate:
220 """Return Data ID associated with this formatter (`DataCoordinate`)."""
221 return self._dataset_ref.dataId
223 @property
224 def dataset_ref(self) -> DatasetRef:
225 """Return Dataset Ref associated with this formatter (`DatasetRef`)."""
226 return self._dataset_ref
228 @property
229 def write_parameters(self) -> Mapping[str, Any]:
230 """Parameters to use when writing out datasets."""
231 if self._write_parameters is not None:
232 return self._write_parameters
233 return {}
235 @property
236 def write_recipes(self) -> Mapping[str, Any]:
237 """Detailed write Recipes indexed by recipe name."""
238 if self._write_recipes is not None:
239 return self._write_recipes
240 return {}
242 def get_write_extension(self) -> str:
243 """Extension to use when writing a file."""
244 default_extension = self.default_extension
245 extension = default_extension if default_extension is not None else ""
246 return extension
248 def can_accept(self, in_memory_dataset: Any) -> bool:
249 """Indicate whether this formatter can accept the specified
250 storage class directly.
252 Parameters
253 ----------
254 in_memory_dataset : `object`
255 The dataset that is to be written.
257 Returns
258 -------
259 accepts : `bool`
260 If `True` the formatter can write data of this type without
261 requiring datastore to convert it. If `False` the datastore
262 will attempt to convert before writing.
264 Notes
265 -----
266 The base class always returns `False` even if the given type is an
267 instance of the storage class type. This will result in a storage
268 class conversion no-op but also allows mocks with mocked storage
269 classes to work properly.
270 """
271 return False
273 @classmethod
274 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
275 """Validate supplied recipes for this formatter.
277 The recipes are supplemented with default values where appropriate.
279 Parameters
280 ----------
281 recipes : `dict`
282 Recipes to validate.
284 Returns
285 -------
286 validated : `dict`
287 Validated recipes.
289 Raises
290 ------
291 RuntimeError
292 Raised if validation fails. The default implementation raises
293 if any recipes are given.
294 """
295 if recipes:
296 raise RuntimeError(f"This formatter does not understand these write recipes: {recipes}")
297 return recipes
299 @classmethod
300 def name(cls) -> str:
301 """Return the fully qualified name of the formatter.
303 Returns
304 -------
305 name : `str`
306 Fully-qualified name of formatter class.
307 """
308 return get_full_type_name(cls)
310 def _is_disassembled(self) -> bool:
311 """Return `True` if this formatter is looking at a disassembled
312 component.
313 """
314 return self.file_descriptor.component is not None
316 def _check_resource_size(self, uri: ResourcePath, recorded_size: int, resource_size: int) -> None:
317 """Compare the recorded size with the resource size.
319 The given URI will not be accessed.
320 """
321 if recorded_size >= 0 and resource_size != recorded_size:
322 raise FileIntegrityError(
323 "Integrity failure in Datastore. "
324 f"Size of file {uri} ({resource_size}) "
325 f"does not match size recorded in registry of {recorded_size}"
326 )
328 def _get_cache_ref(self) -> DatasetRef:
329 """Get the `DatasetRef` to use for cache look ups.
331 Returns
332 -------
333 ref : `lsst.daf.butler.DatasetRef`
334 The dataset ref to use when looking in the cache.
335 For single-file dataset this will be the dataset ref directly.
336 If this is disassembled we need the component and the component
337 will be in the `FileDescriptor`.
338 """
339 if self.file_descriptor.component is None:
340 cache_ref = self.dataset_ref
341 else:
342 cache_ref = self.dataset_ref.makeComponentRef(self.file_descriptor.component)
343 return cache_ref
345 def _ensure_cache(
346 self, cache_manager: AbstractDatastoreCacheManager | None = None
347 ) -> AbstractDatastoreCacheManager:
348 """Return the cache if given else return a null cache."""
349 if cache_manager is None:
350 # Circular import avoidance.
351 from .datastore.cache_manager import DatastoreDisabledCacheManager
353 cache_manager = DatastoreDisabledCacheManager(None, None)
354 return cache_manager
356 def read(
357 self,
358 component: str | None = None,
359 expected_size: int = -1,
360 cache_manager: AbstractDatastoreCacheManager | None = None,
361 ) -> Any:
362 """Read a Dataset.
364 Parameters
365 ----------
366 component : `str`, optional
367 Component to read from the file. Only used if the `StorageClass`
368 for reading differed from the `StorageClass` used to write the
369 file.
370 expected_size : `int`, optional
371 If known, the expected size of the resource to read. This can be
372 used for verification or to decide whether to do a direct read or a
373 file download. ``-1`` indicates the file size is not known.
374 cache_manager : `AbstractDatastoreCacheManager`
375 A cache manager to use to allow a formatter to cache a remote file
376 locally or read a cached file that is already local.
378 Returns
379 -------
380 in_memory_dataset : `object`
381 The requested Dataset.
383 Notes
384 -----
385 This method should not be subclassed. Instead formatter subclasses
386 should re-implement the specific `read_from_*` methods as appropriate.
387 Each of these methods has a corresponding class property that must
388 be `True` for the method to be called.
390 The priority for reading is:
392 * `read_from_uri`
393 * `read_from_stream`
394 * `read_from_local_file`
395 * `read_from_uri` (but with a local file)
397 Any of these methods can return `NotImplemented` if there is a desire
398 to skip to the next one in the list. If a dataset is being requested
399 with no component, no parameters, and it should also be added to the
400 local cache, the first two calls will be skipped (unless
401 `read_from_stream` is the only implemented read method) such that a
402 local file will be used.
404 A Formatter can also read a file from within a Zip file if the
405 URI associated with the `FileDescriptor` corresponds to a file with
406 a `.zip` extension and a URI fragment of the form
407 ``zip-path={path_in_zip}``. When reading a file from within a Zip
408 file the priority for reading is:
410 * `read_from_stream`
411 * `read_from_local_file`
412 * `read_from_uri`
414 There are multiple cases that must be handled for reading:
416 For a single file:
418 * No component requested, read the whole file.
419 * Component requested, optionally read the component efficiently,
420 else read the whole file and extract the component.
421 * Derived component requested, read whole file or read relevant
422 component and derive.
424 Disassembled Composite:
426 * The file to read here is the component itself. Formatter only knows
427 about this one component file. Should be no component specified
428 in the ``read`` call but the `FileDescriptor` will know which
429 component this is.
430 * A derived component. The file to read is a component but not the
431 specified component. The caching needs the component from which
432 it's derived.
434 Raises
435 ------
436 FormatterNotImplementedError
437 Raised if no implementations were found that could read this
438 resource.
439 """
440 # If the file to read is a ZIP file with a fragment requesting
441 # a file within the ZIP file, it is no longer possible to use the
442 # direct read from URI option and the contents of the Zip file must
443 # be extracted.
444 uri = self.file_descriptor.location.uri
445 if uri.fragment and uri.unquoted_fragment.startswith("zip-path="):
446 _, _, path_in_zip = uri.unquoted_fragment.partition("=")
448 # Open the Zip file using ResourcePath.
449 with uri.open("rb") as fd:
450 with zipfile.ZipFile(fd) as zf: # type: ignore
451 if self.can_read_from_stream:
452 with contextlib.closing(zf.open(path_in_zip)) as zip_fd:
453 result = self.read_from_stream(zip_fd, component, expected_size=expected_size)
455 if result is not NotImplemented:
456 return result
458 # For now for both URI and local file options we retrieve
459 # the bytes to a temporary local and use that.
460 _, suffix = os.path.splitext(path_in_zip)
461 with ResourcePath.temporary_uri(suffix=suffix) as tmp_uri:
462 tmp_uri.write(zf.read(path_in_zip))
464 if self.can_read_from_local_file:
465 result = self.read_from_local_file(
466 tmp_uri.ospath, component, expected_size=expected_size
467 )
468 if result is not NotImplemented:
469 return result
470 if self.can_read_from_uri:
471 result = self.read_from_uri(tmp_uri, component, expected_size=expected_size)
472 if result is not NotImplemented:
473 return result
475 raise FormatterNotImplementedError(
476 f"Formatter {self.name()} could not read the file at {uri} using any method."
477 )
479 # If the there are no parameters, no component request, and
480 # the file should be cached, it is better to defer the read_from_uri
481 # to use the local file and populate the cache.
482 prefer_local = False
483 if (
484 component is None
485 and not self.file_descriptor.parameters
486 and self._ensure_cache(cache_manager).should_be_cached(self._get_cache_ref())
487 ):
488 prefer_local = True
490 # First see if the formatter can support direct remote read from
491 # a URI. This can be called later from the local path. If it returns
492 # NotImplemented the formatter decided on its own that local
493 # reads are preferred.
494 if not prefer_local and self.can_read_from_uri:
495 result = self.read_directly_from_possibly_cached_uri(
496 component, expected_size, cache_manager=cache_manager
497 )
498 if result is not NotImplemented:
499 return result
501 # Some formatters might want to be able to read directly from
502 # an open file stream. This is preferred over forcing a download
503 # to local file system unless a file read option is available and we
504 # want to store it in the cache because the whole file is being read.
505 if self.can_read_from_stream and not (
506 prefer_local and (self.can_read_from_uri or self.can_read_from_local_file)
507 ):
508 result = self.read_from_possibly_cached_stream(
509 component, expected_size, cache_manager=cache_manager
510 )
511 if result is not NotImplemented:
512 return result
514 # Finally, try to read the local file.
515 if self.can_read_from_local_file or self.can_read_from_uri:
516 result = self.read_from_possibly_cached_local_file(
517 component, expected_size, cache_manager=cache_manager
518 )
519 if result is not NotImplemented:
520 return result
522 raise FormatterNotImplementedError(
523 f"Formatter {self.name()} could not read the file at {uri} using any method."
524 )
526 def _read_from_possibly_cached_location_no_cache_write(
527 self,
528 callback: Callable[[ResourcePath, str | None, int], Any],
529 component: str | None = None,
530 expected_size: int = -1,
531 *,
532 cache_manager: AbstractDatastoreCacheManager | None = None,
533 ) -> Any:
534 """Read from the cache and call payload without writing to cache."""
535 cache_manager = self._ensure_cache(cache_manager)
537 uri = self.file_descriptor.location.uri
538 cache_ref = self._get_cache_ref()
540 # The component for log messages is either the component requested
541 # explicitly or the component from the file descriptor.
542 log_component = component if component is not None else self.file_descriptor.component
544 with cache_manager.find_in_cache(cache_ref, uri.getExtension()) as cached_file:
545 if cached_file is not None:
546 desired_uri = cached_file
547 msg = f" (cached version of {uri})"
548 else:
549 desired_uri = uri
550 msg = ""
552 if desired_uri.isLocal:
553 # Do not spend the time doing a slow size() call to a remote
554 # resource.
555 self._check_resource_size(desired_uri, expected_size, desired_uri.size())
557 with time_this(
558 log,
559 msg="Reading%s from file handle %s%s with formatter %s",
560 args=(
561 f" component {log_component}" if log_component else "",
562 desired_uri,
563 msg,
564 self.name(),
565 ),
566 ):
567 return callback(desired_uri, component, expected_size)
569 def read_from_possibly_cached_stream(
570 self,
571 component: str | None = None,
572 expected_size: int = -1,
573 *,
574 cache_manager: AbstractDatastoreCacheManager | None = None,
575 ) -> Any:
576 """Read from a stream, checking for possible presence in local cache.
578 Parameters
579 ----------
580 component : `str`, optional
581 Component to read from the file. Only used if the `StorageClass`
582 for reading differed from the `StorageClass` used to write the
583 file.
584 expected_size : `int`, optional
585 If known, the expected size of the resource to read. This can be
586 used for verification or to decide whether to do a direct read or a
587 file download. ``-1`` indicates the file size is not known.
588 cache_manager : `AbstractDatastoreCacheManager`
589 A cache manager to use to allow a formatter to check if there is
590 a copy of the file in the local cache.
592 Returns
593 -------
594 in_memory_dataset : `object` or `NotImplemented`
595 The requested Dataset or an indication that the read mode was
596 not implemented.
598 Notes
599 -----
600 Calls `read_from_stream` but will first check the datastore cache
601 in case the file is present locally. This method will not download
602 a file to the local cache.
603 """
605 def _open_stream(uri: ResourcePath, comp: str | None, size: int = -1) -> Any:
606 with uri.open("rb") as fd:
607 return self.read_from_stream(fd, comp, expected_size=size)
609 return self._read_from_possibly_cached_location_no_cache_write(
610 _open_stream, component, expected_size=expected_size, cache_manager=cache_manager
611 )
613 def read_directly_from_possibly_cached_uri(
614 self,
615 component: str | None = None,
616 expected_size: int = -1,
617 *,
618 cache_manager: AbstractDatastoreCacheManager | None = None,
619 ) -> Any:
620 """Read from arbitrary URI, checking for possible presence in local
621 cache.
623 Parameters
624 ----------
625 component : `str`, optional
626 Component to read from the file. Only used if the `StorageClass`
627 for reading differed from the `StorageClass` used to write the
628 file.
629 expected_size : `int`, optional
630 If known, the expected size of the resource to read. This can be
631 ``-1`` indicates the file size is not known.
632 cache_manager : `AbstractDatastoreCacheManager`
633 A cache manager to use to allow a formatter to check if there is
634 a copy of the file in the local cache.
636 Returns
637 -------
638 in_memory_dataset : `object` or `NotImplemented`
639 The requested Dataset or an indication that the read mode was
640 not implemented.
642 Notes
643 -----
644 This method will first check the datastore cache
645 in case the file is present locally. This method will not cache a
646 remote dataset and will only do a size check for local files to avoid
647 unnecessary round trips to a remote server.
649 The URI will be read by calling `read_from_uri`.
650 """
652 def _open_uri(uri: ResourcePath, comp: str | None, size: int = -1) -> Any:
653 return self.read_from_uri(uri, comp, expected_size=size)
655 return self._read_from_possibly_cached_location_no_cache_write(
656 _open_uri, component, expected_size=expected_size, cache_manager=cache_manager
657 )
659 def read_from_possibly_cached_local_file(
660 self,
661 component: str | None = None,
662 expected_size: int = -1,
663 *,
664 cache_manager: AbstractDatastoreCacheManager | None = None,
665 ) -> Any:
666 """Read a dataset ensuring that a local file is used, checking the
667 cache for it.
669 Parameters
670 ----------
671 component : `str`, optional
672 Component to read from the file. Only used if the `StorageClass`
673 for reading differed from the `StorageClass` used to write the
674 file.
675 expected_size : `int`, optional
676 If known, the expected size of the resource to read. This can be
677 used for verification or to decide whether to do a direct read or a
678 file download. ``-1`` indicates the file size is not known.
679 cache_manager : `AbstractDatastoreCacheManager`
680 A cache manager to use to allow a formatter to cache a remote file
681 locally or read a cached file that is already local.
683 Returns
684 -------
685 in_memory_dataset : `object` or `NotImplemented`
686 The requested Dataset or an indication that the read mode was
687 not implemented.
689 Notes
690 -----
691 The file will be downloaded and cached if it is a remote resource.
692 The file contents will be read using `read_from_local_file` or
693 `read_from_uri`, with preference given to the former.
694 """
695 cache_manager = self._ensure_cache(cache_manager)
696 uri = self.file_descriptor.location.uri
698 # Need to have something we can look up in the cache.
699 cache_ref = self._get_cache_ref()
701 # The component for log messages is either the component requested
702 # explicitly or the component from the file descriptor.
703 log_component = component if component is not None else self.file_descriptor.component
705 result = NotImplemented
707 # Ensure we have a local file.
708 with cache_manager.find_in_cache(cache_ref, uri.getExtension()) as cached_file:
709 if cached_file is not None:
710 msg = f"(via cache read of remote file {uri})"
711 uri = cached_file
712 else:
713 msg = ""
715 with uri.as_local() as local_uri:
716 self._check_resource_size(self.file_descriptor.location.uri, expected_size, local_uri.size())
717 can_be_cached = False
718 if uri != local_uri:
719 # URI was remote and file was downloaded
720 cache_msg = ""
722 if cache_manager.should_be_cached(cache_ref):
723 # In this scenario we want to ask if the downloaded
724 # file should be cached but we should not cache
725 # it until after we've used it (to ensure it can't
726 # be expired whilst we are using it).
727 can_be_cached = True
729 # Say that it is "likely" to be cached because
730 # if the formatter read fails we will not be
731 # caching this file.
732 cache_msg = " and likely cached"
734 msg = f"(via download to local file{cache_msg})"
736 with time_this(
737 log,
738 msg="Reading%s from location %s %s with formatter %s",
739 args=(
740 f" component {log_component}" if log_component else "",
741 uri,
742 msg,
743 self.name(),
744 ),
745 ):
746 if self.can_read_from_local_file:
747 result = self.read_from_local_file(
748 local_uri.ospath, component=component, expected_size=expected_size
749 )
750 if result is NotImplemented and self.can_read_from_uri:
751 # If the direct URI reader was skipped earlier and
752 # there is no explicit local file implementation, pass
753 # in the guaranteed local URI to the generic reader.
754 result = self.read_from_uri(
755 local_uri, component=component, expected_size=expected_size
756 )
758 # File was read successfully so can move to cache.
759 # Also move to cache even if NotImplemented was returned.
760 if can_be_cached:
761 cache_manager.move_to_cache(local_uri, cache_ref)
763 return result
765 def read_from_uri(self, uri: ResourcePath, component: str | None = None, expected_size: int = -1) -> Any:
766 """Read a dataset from a URI that can be local or remote.
768 Parameters
769 ----------
770 uri : `lsst.resources.ResourcePath`
771 URI to use to read the dataset. This URI can be local or remote
772 and can refer to the actual resource or to a locally cached file.
773 component : `str` or `None`, optional
774 The component to be read from the dataset.
775 expected_size : `int`, optional
776 If known, the expected size of the resource to read. This can be
777 ``-1`` indicates the file size is not known.
779 Returns
780 -------
781 in_memory_dataset : `object` or `NotImplemented`
782 The Python object read from the resource or `NotImplemented`.
784 Raises
785 ------
786 FormatterNotImplementedError
787 Raised if there is no support for direct reads from a, possibly,
788 remote URI.
790 Notes
791 -----
792 This method is only called if the class property
793 ``can_read_from_uri`` is set to `True`.
795 It is possible that a cached local file will be given to this method
796 even if it was originally a remote URI. This can happen if the original
797 write resulted in the file being added to the local cache.
799 If the full file is being read this file will not be added to the
800 local cache. Consider returning `NotImplemented` in
801 this situation, for example if there are no parameters or component
802 specified, and allowing the system to fall back to calling
803 `read_from_local_file` (which will populate the cache if configured
804 to do so).
805 """
806 return NotImplemented
808 def read_from_stream(
809 self, stream: BinaryIO | ResourceHandleProtocol, component: str | None = None, expected_size: int = -1
810 ) -> Any:
811 """Read from an open file descriptor.
813 Parameters
814 ----------
815 stream : `lsst.resources.ResourceHandleProtocol` or \
816 `typing.BinaryIO`
817 File stream to use to read the dataset.
818 component : `str` or `None`, optional
819 The component to be read from the dataset.
820 expected_size : `int`, optional
821 If known, the expected size of the resource to read. This can be
822 ``-1`` indicates the file size is not known.
824 Returns
825 -------
826 in_memory_dataset : `object` or `NotImplemented`
827 The Python object read from the stream or `NotImplemented`.
829 Notes
830 -----
831 Only called if the class property ``can_read_from_stream`` is `True`.
832 """
833 return NotImplemented
835 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any:
836 """Read a dataset from a URI guaranteed to refer to the local file
837 system.
839 Parameters
840 ----------
841 path : `str`
842 Path to a local file that should be read.
843 component : `str` or `None`, optional
844 The component to be read from the dataset.
845 expected_size : `int`, optional
846 If known, the expected size of the resource to read. This can be
847 ``-1`` indicates the file size is not known.
849 Returns
850 -------
851 in_memory_dataset : `object` or `NotImplemented`
852 The Python object read from the resource or `NotImplemented`.
854 Raises
855 ------
856 FormatterNotImplementedError
857 Raised if there is no implementation written to read data
858 from a local file.
860 Notes
861 -----
862 This method will only be called if the class property
863 ``can_read_from_local_file`` is `True` and other options were not
864 used.
865 """
866 return NotImplemented
868 def add_provenance(
869 self, in_memory_dataset: Any, /, *, provenance: DatasetProvenance | None = None
870 ) -> Any:
871 """Add provenance to the dataset.
873 Parameters
874 ----------
875 in_memory_dataset : `object`
876 The dataset to serialize.
877 provenance : `DatasetProvenance` or `None`, optional
878 Provenance to attach to dataset.
880 Returns
881 -------
882 dataset_to_write : `object`
883 The dataset to use for serialization. Can be the same object as
884 given.
886 Notes
887 -----
888 The base class implementation returns the given object unchanged.
889 """
890 return in_memory_dataset
892 @final
893 def write(
894 self,
895 in_memory_dataset: Any,
896 /,
897 *,
898 cache_manager: AbstractDatastoreCacheManager | None = None,
899 provenance: DatasetProvenance | None = None,
900 ) -> None:
901 """Write a Dataset.
903 Parameters
904 ----------
905 in_memory_dataset : `object`
906 The Dataset to serialize.
907 cache_manager : `AbstractDatastoreCacheManager`
908 A cache manager to use to allow a formatter to cache the written
909 file.
910 provenance : `DatasetProvenance` | `None`, optional
911 Provenance to attach to the file being written.
913 Returns
914 -------
915 None
917 Raises
918 ------
919 FormatterNotImplementedError
920 Raised if the formatter subclass has not implemented
921 `write_local_file` and `to_bytes` was not called.
922 Exception
923 Raised if there is an error serializing the dataset to disk.
925 Notes
926 -----
927 The intent is for subclasses to implement either `to_bytes` or
928 `write_local_file` or both and not to subclass this method.
929 """
930 # Ensure we are using the correct file extension.
931 uri = self.file_descriptor.location.uri.updatedExtension(self.get_write_extension())
933 # Attach any provenance to the dataset. This could involve returning
934 # a different object.
935 in_memory_dataset = self.add_provenance(in_memory_dataset, provenance=provenance)
937 written = self.write_direct(in_memory_dataset, uri, cache_manager)
938 if not written:
939 self.write_locally_then_move(in_memory_dataset, uri, cache_manager)
941 def write_direct(
942 self,
943 in_memory_dataset: Any,
944 uri: ResourcePath,
945 cache_manager: AbstractDatastoreCacheManager | None = None,
946 ) -> bool:
947 """Serialize and write directly to final location.
949 Parameters
950 ----------
951 in_memory_dataset : `object`
952 The Dataset to serialize.
953 uri : `lsst.resources.ResourcePath`
954 URI to use when writing the serialized dataset.
955 cache_manager : `AbstractDatastoreCacheManager`
956 A cache manager to use to allow a formatter to cache the written
957 file.
959 Returns
960 -------
961 written : `bool`
962 Flag to indicate whether the direct write did happen.
964 Raises
965 ------
966 Exception
967 Raised if there was a failure from serializing to bytes that
968 was not `FormatterNotImplementedError`.
970 Notes
971 -----
972 This method will call `to_bytes` to serialize the in-memory dataset
973 and then will call the `~lsst.resources.ResourcePath.write` method
974 directly.
976 If the dataset should be cached or is local the file will not be
977 written and the method will return `False`. This is because local URIs
978 should be written to a temporary file name and then renamed to allow
979 atomic writes. That path is handled by `write_locally_then_move`
980 through `write_local_file`) and is preferred over this method being
981 subclassed and the atomic write re-implemented.
982 """
983 cache_manager = self._ensure_cache(cache_manager)
985 # For remote URIs some datasets can be serialized directly
986 # to bytes and sent to the remote datastore without writing a
987 # file. If the dataset is intended to be saved to the cache
988 # a file is always written and direct write to the remote
989 # datastore is bypassed.
990 data_written = False
991 if not uri.isLocal and not cache_manager.should_be_cached(self._get_cache_ref()):
992 # Remote URI that is not cached so can write directly.
993 try:
994 serialized_dataset = self.to_bytes(in_memory_dataset)
995 except FormatterNotImplementedError:
996 # Fallback to the file writing option.
997 pass
998 except Exception as e:
999 e.add_note(
1000 f"Failed to serialize dataset {self.dataset_ref} of "
1001 f"type {get_full_type_name(in_memory_dataset)} to bytes."
1002 )
1003 raise
1004 else:
1005 log.debug("Writing bytes directly to %s", uri)
1006 uri.write(serialized_dataset, overwrite=True)
1007 log.debug("Successfully wrote bytes directly to %s", uri)
1008 data_written = True
1009 return data_written
1011 def write_locally_then_move(
1012 self,
1013 in_memory_dataset: Any,
1014 uri: ResourcePath,
1015 cache_manager: AbstractDatastoreCacheManager | None = None,
1016 ) -> None:
1017 """Write file to file system and then move to final location.
1019 Parameters
1020 ----------
1021 in_memory_dataset : `object`
1022 The Dataset to serialize.
1023 uri : `lsst.resources.ResourcePath`
1024 URI to use when writing the serialized dataset.
1025 cache_manager : `AbstractDatastoreCacheManager`
1026 A cache manager to use to allow a formatter to cache the written
1027 file.
1029 Raises
1030 ------
1031 FormatterNotImplementedError
1032 Raised if the formatter subclass has not implemented
1033 `write_local_file`.
1034 Exception
1035 Raised if there is an error serializing the dataset to disk.
1036 """
1037 cache_manager = self._ensure_cache(cache_manager)
1039 with TemporaryForIngest.make_path(uri) as temporary_uri:
1040 # Need to configure the formatter to write to a different
1041 # location and that needs us to overwrite internals
1042 log.debug("Writing dataset to temporary location at %s", temporary_uri)
1044 # Assumes that if write_local_file is not subclassed that
1045 # to_bytes will be called using the base class definition.
1046 try:
1047 self.write_local_file(in_memory_dataset, temporary_uri)
1048 except Exception as e:
1049 e.add_note(
1050 f"Failed to serialize dataset {self.dataset_ref} of type"
1051 f" {get_full_type_name(in_memory_dataset)} to "
1052 f"temporary location {temporary_uri}."
1053 )
1054 raise
1056 # Use move for a local file since that becomes an efficient
1057 # os.rename. For remote resources we use copy to allow the
1058 # file to be cached afterwards.
1059 transfer = "move" if uri.isLocal else "copy"
1061 uri.transfer_from(temporary_uri, transfer=transfer, overwrite=True)
1063 if transfer == "copy":
1064 # Cache if required
1065 cache_manager.move_to_cache(temporary_uri, self._get_cache_ref())
1067 log.debug("Successfully wrote dataset to %s via a temporary file.", uri)
1069 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
1070 """Serialize the in-memory dataset to a local file.
1072 Parameters
1073 ----------
1074 in_memory_dataset : `object`
1075 The Python object to serialize.
1076 uri : `ResourcePath`
1077 The URI to use when writing the file.
1079 Notes
1080 -----
1081 By default this method will attempt to call `to_bytes` and then
1082 write these bytes to the file.
1084 Raises
1085 ------
1086 FormatterNotImplementedError
1087 Raised if the formatter subclass has not implemented this method
1088 or has failed to implement the `to_bytes` method.
1089 """
1090 log.debug("Writing bytes directly to %s", uri)
1091 uri.write(self.to_bytes(in_memory_dataset))
1092 log.debug("Successfully wrote bytes directly to %s", uri)
1094 def to_bytes(self, in_memory_dataset: Any) -> bytes:
1095 """Serialize the in-memory dataset to bytes.
1097 Parameters
1098 ----------
1099 in_memory_dataset : `object`
1100 The Python object to serialize.
1102 Returns
1103 -------
1104 serialized_dataset : `bytes`
1105 Bytes representing the serialized dataset.
1107 Raises
1108 ------
1109 FormatterNotImplementedError
1110 Raised if the formatter has not implemented the method. This will
1111 not cause a problem if `write_local_file` has been implemented.
1112 """
1113 raise FormatterNotImplementedError(
1114 f"This formatter can not convert {get_full_type_name(in_memory_dataset)} directly to bytes."
1115 )
1117 def make_updated_location(self, location: Location) -> Location:
1118 """Return a new `Location` updated with this formatter's extension.
1120 Parameters
1121 ----------
1122 location : `Location`
1123 The location to update.
1125 Returns
1126 -------
1127 updated : `Location`
1128 A new `Location` with a new file extension applied.
1129 """
1130 location = location.clone()
1131 # If the extension is "" the extension will be removed.
1132 location.updateExtension(self.get_write_extension())
1133 return location
1135 @classmethod
1136 def validate_extension(cls, location: Location) -> None:
1137 """Check the extension of the provided location for compatibility.
1139 Parameters
1140 ----------
1141 location : `Location`
1142 Location from which to extract a file extension.
1144 Returns
1145 -------
1146 None
1148 Raises
1149 ------
1150 ValueError
1151 Raised if the formatter does not understand this extension.
1152 """
1153 supported = set(cls.supported_extensions)
1154 default = cls.default_extension # type: ignore
1156 # If extension is implemented as an instance property it won't return
1157 # a string when called as a class property. Assume that
1158 # the supported extensions class property is complete.
1159 if default is not None and isinstance(default, str):
1160 supported.add(default)
1162 # Get the file name from the uri
1163 file = location.uri.basename()
1165 # Check that this file name ends with one of the supported extensions.
1166 # This is less prone to confusion than asking the location for
1167 # its extension and then doing a set comparison
1168 for ext in supported:
1169 if file.endswith(ext):
1170 return
1172 raise ValueError(
1173 f"Extension '{location.getExtension()}' on '{location}' "
1174 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})"
1175 )
1177 def predict_path(self) -> str:
1178 """Return the path that would be returned by write.
1180 Does not write any data file.
1182 Uses the `FileDescriptor` associated with the instance.
1184 Returns
1185 -------
1186 path : `str`
1187 Path within datastore that would be associated with the location
1188 stored in this `Formatter`.
1189 """
1190 updated = self.make_updated_location(self.file_descriptor.location)
1191 return updated.pathInStore.path
1193 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
1194 """Segregate the supplied parameters.
1196 This splits the parameters into those understood by the
1197 formatter and those not understood by the formatter.
1199 Any unsupported parameters are assumed to be usable by associated
1200 assemblers.
1202 Parameters
1203 ----------
1204 parameters : `dict`, optional
1205 Parameters with values that have been supplied by the caller
1206 and which might be relevant for the formatter. If `None`
1207 parameters will be read from the registered `FileDescriptor`.
1209 Returns
1210 -------
1211 supported : `dict`
1212 Those parameters supported by this formatter.
1213 unsupported : `dict`
1214 Those parameters not supported by this formatter.
1215 """
1216 if parameters is None:
1217 parameters = self.file_descriptor.parameters
1219 if parameters is None:
1220 return {}, {}
1222 if self.unsupported_parameters is None:
1223 # Support none of the parameters
1224 return {}, parameters.copy()
1226 # Start by assuming all are supported
1227 supported = parameters.copy()
1228 unsupported = {}
1230 # And remove any we know are not supported
1231 for p in set(supported):
1232 if p in self.unsupported_parameters:
1233 unsupported[p] = supported.pop(p)
1235 return supported, unsupported
1238class Formatter(metaclass=ABCMeta):
1239 """Interface for reading and writing Datasets.
1241 The formatters are associated with a particular `StorageClass`.
1243 Parameters
1244 ----------
1245 fileDescriptor : `FileDescriptor`, optional
1246 Identifies the file to read or write, and the associated storage
1247 classes and parameter information.
1248 dataId : `DataCoordinate`
1249 Data ID associated with this formatter.
1250 writeParameters : `dict`, optional
1251 Parameters to control how the dataset is serialized.
1252 writeRecipes : `dict`, optional
1253 Detailed write recipes indexed by recipe name.
1254 **kwargs
1255 Additional parameters that can allow parameters
1256 from `FormatterV2` to be provided.
1258 Notes
1259 -----
1260 All Formatter subclasses should share the base class's constructor
1261 signature.
1262 """
1264 # Now assuming that Formatter v1 can only refer to files so can add
1265 # this property for compatibility with v2.
1266 extension: str | None = None
1267 """Default file extension to use for writing files. None means that no
1268 modifications will be made to the supplied file extension. (`str`)"""
1270 unsupportedParameters: ClassVar[Set[str] | None] = frozenset()
1271 """Set of read parameters not understood by this `Formatter`. An empty set
1272 means all parameters are supported. `None` indicates that no parameters
1273 are supported. These parameters should match those defined in the storage
1274 class definition. (`frozenset`).
1275 """
1277 supportedWriteParameters: ClassVar[Set[str] | None] = None
1278 """Parameters understood by this formatter that can be used to control
1279 how a dataset is serialized. `None` indicates that no parameters are
1280 supported."""
1282 supportedExtensions: ClassVar[Set[str]] = frozenset()
1283 """Set of all extensions supported by this formatter.
1285 Only expected to be populated by Formatters that write files. Any extension
1286 assigned to the ``extension`` property will be automatically included in
1287 the list of supported extensions."""
1289 def __init__(
1290 self,
1291 fileDescriptor: FileDescriptor,
1292 *,
1293 dataId: DataCoordinate | None = None,
1294 writeParameters: Mapping[str, Any] | None = None,
1295 writeRecipes: Mapping[str, Any] | None = None,
1296 # Allow FormatterV2 parameters to be dropped.
1297 **kwargs: Any,
1298 ):
1299 if not isinstance(fileDescriptor, FileDescriptor):
1300 raise TypeError("File descriptor must be a FileDescriptor")
1301 self._fileDescriptor = fileDescriptor
1303 if dataId is None:
1304 raise RuntimeError("dataId is now required for formatter initialization")
1305 if not isinstance(dataId, DataCoordinate):
1306 raise TypeError(f"DataId is required to be a DataCoordinate but got {type(dataId)}.")
1307 self._dataId = dataId
1309 # V2 compatibility.
1310 if writeParameters is None:
1311 writeParameters = kwargs.get("write_parameters")
1312 if writeRecipes is None:
1313 writeRecipes = kwargs.get("write_recipes")
1315 # Check that the write parameters are allowed
1316 if writeParameters:
1317 if self.supportedWriteParameters is None:
1318 raise ValueError(
1319 f"This formatter does not accept any write parameters. Got: {', '.join(writeParameters)}"
1320 )
1321 else:
1322 given = set(writeParameters)
1323 unknown = given - self.supportedWriteParameters
1324 if unknown:
1325 s = "s" if len(unknown) != 1 else ""
1326 unknownStr = ", ".join(f"'{u}'" for u in unknown)
1327 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}")
1329 self._writeParameters = writeParameters
1330 self._writeRecipes = self.validate_write_recipes(writeRecipes)
1332 def __str__(self) -> str:
1333 return f"{self.name()}@{self.fileDescriptor.location.path}"
1335 def __repr__(self) -> str:
1336 return f"{self.name()}({self.fileDescriptor!r})"
1338 @property
1339 def fileDescriptor(self) -> FileDescriptor:
1340 """File descriptor associated with this formatter
1341 (`FileDescriptor`).
1342 """
1343 return self._fileDescriptor
1345 @property
1346 def dataId(self) -> DataCoordinate:
1347 """Return Data ID associated with this formatter (`DataCoordinate`)."""
1348 return self._dataId
1350 @property
1351 def writeParameters(self) -> Mapping[str, Any]:
1352 """Parameters to use when writing out datasets."""
1353 if self._writeParameters is not None:
1354 return self._writeParameters
1355 return {}
1357 @property
1358 def writeRecipes(self) -> Mapping[str, Any]:
1359 """Detailed write Recipes indexed by recipe name."""
1360 if self._writeRecipes is not None:
1361 return self._writeRecipes
1362 return {}
1364 def can_accept(self, in_memory_dataset: Any) -> bool:
1365 """Indicate whether this formatter can accept the specified
1366 storage class directly.
1368 Parameters
1369 ----------
1370 in_memory_dataset : `object`
1371 The dataset that is to be written.
1373 Returns
1374 -------
1375 accepts : `bool`
1376 If `True` the formatter can write data of this type without
1377 requiring datastore to convert it. If `False` the datastore
1378 will attempt to convert before writing.
1380 Notes
1381 -----
1382 The base class checks that the given python type matches
1383 the python type specified for this formatter when
1384 constructed.
1385 """
1386 return isinstance(in_memory_dataset, self.file_descriptor.storageClass.pytype)
1388 @classmethod
1389 def validateWriteRecipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
1390 """Validate supplied recipes for this formatter.
1392 The recipes are supplemented with default values where appropriate.
1394 Parameters
1395 ----------
1396 recipes : `dict`
1397 Recipes to validate.
1399 Returns
1400 -------
1401 validated : `dict`
1402 Validated recipes.
1404 Raises
1405 ------
1406 RuntimeError
1407 Raised if validation fails. The default implementation raises
1408 if any recipes are given.
1409 """
1410 if recipes:
1411 raise RuntimeError(f"This formatter does not understand these writeRecipes: {recipes}")
1412 return recipes
1414 @classmethod
1415 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
1416 return cls.validateWriteRecipes(recipes)
1418 @classmethod
1419 def name(cls) -> str:
1420 """Return the fully qualified name of the formatter.
1422 Returns
1423 -------
1424 name : `str`
1425 Fully-qualified name of formatter class.
1426 """
1427 return get_full_type_name(cls)
1429 @abstractmethod
1430 def read(self, component: str | None = None) -> Any:
1431 """Read a Dataset.
1433 Parameters
1434 ----------
1435 component : `str`, optional
1436 Component to read from the file. Only used if the `StorageClass`
1437 for reading differed from the `StorageClass` used to write the
1438 file.
1440 Returns
1441 -------
1442 inMemoryDataset : `object`
1443 The requested Dataset.
1444 """
1445 raise FormatterNotImplementedError("Type does not support reading")
1447 @abstractmethod
1448 def write(self, inMemoryDataset: Any) -> None:
1449 """Write a Dataset.
1451 Parameters
1452 ----------
1453 inMemoryDataset : `object`
1454 The Dataset to store.
1455 """
1456 raise FormatterNotImplementedError("Type does not support writing")
1458 @classmethod
1459 def can_read_bytes(cls) -> bool:
1460 """Indicate if this formatter can format from bytes.
1462 Returns
1463 -------
1464 can : `bool`
1465 `True` if the `fromBytes` method is implemented.
1466 """
1467 # We have no property to read so instead try to format from a byte
1468 # and see what happens
1469 try:
1470 # We know the arguments are incompatible
1471 cls.fromBytes(cls, b"") # type: ignore
1472 except FormatterNotImplementedError:
1473 return False
1474 except Exception:
1475 # There will be problems with the bytes we are supplying so ignore
1476 pass
1477 return True
1479 def fromBytes(self, serializedDataset: bytes, component: str | None = None) -> object:
1480 """Read serialized data into a Dataset or its component.
1482 Parameters
1483 ----------
1484 serializedDataset : `bytes`
1485 Bytes object to unserialize.
1486 component : `str`, optional
1487 Component to read from the Dataset. Only used if the `StorageClass`
1488 for reading differed from the `StorageClass` used to write the
1489 file.
1491 Returns
1492 -------
1493 inMemoryDataset : `object`
1494 The requested data as a Python object. The type of object
1495 is controlled by the specific formatter.
1496 """
1497 raise FormatterNotImplementedError("Type does not support reading from bytes.")
1499 def toBytes(self, inMemoryDataset: Any) -> bytes:
1500 """Serialize the Dataset to bytes based on formatter.
1502 Parameters
1503 ----------
1504 inMemoryDataset : `object`
1505 The Python object to serialize.
1507 Returns
1508 -------
1509 serializedDataset : `bytes`
1510 Bytes representing the serialized dataset.
1511 """
1512 raise FormatterNotImplementedError("Type does not support writing to bytes.")
1514 @contextlib.contextmanager
1515 def _updateLocation(self, location: Location | None) -> Iterator[Location]:
1516 """Temporarily replace the location associated with this formatter.
1518 Parameters
1519 ----------
1520 location : `Location`
1521 New location to use for this formatter. If `None` the
1522 formatter will not change but it will still return
1523 the old location. This allows it to be used in a code
1524 path where the location may not need to be updated
1525 but the with block is still convenient.
1527 Yields
1528 ------
1529 old : `Location`
1530 The old location that will be restored.
1532 Notes
1533 -----
1534 This is an internal method that should be used with care.
1535 It may change in the future. Should be used as a context
1536 manager to restore the location when the temporary is no
1537 longer required.
1538 """
1539 old = self._fileDescriptor.location
1540 try:
1541 if location is not None:
1542 self._fileDescriptor.location = location
1543 yield old
1544 finally:
1545 if location is not None:
1546 self._fileDescriptor.location = old
1548 def make_updated_location(self, location: Location) -> Location:
1549 """Return a new `Location` updated with this formatter's extension.
1551 Parameters
1552 ----------
1553 location : `Location`
1554 The location to update.
1556 Returns
1557 -------
1558 updated : `Location`
1559 A new `Location` with a new file extension applied.
1561 Raises
1562 ------
1563 NotImplementedError
1564 Raised if there is no ``extension`` attribute associated with
1565 this formatter.
1567 Notes
1568 -----
1569 This method is available to all Formatters but might not be
1570 implemented by all formatters. It requires that a formatter set
1571 an ``extension`` attribute containing the file extension used when
1572 writing files. If ``extension`` is `None` the supplied file will
1573 not be updated. Not all formatters write files so this is not
1574 defined in the base class.
1575 """
1576 location = location.clone()
1577 try:
1578 # We are deliberately allowing extension to be undefined by
1579 # default in the base class and mypy complains.
1580 location.updateExtension(self.extension) # type:ignore
1581 except AttributeError:
1582 raise NotImplementedError("No file extension registered with this formatter") from None
1583 return location
1585 @classmethod
1586 def validate_extension(cls, location: Location) -> None:
1587 """Check the extension of the provided location for compatibility.
1589 Parameters
1590 ----------
1591 location : `Location`
1592 Location from which to extract a file extension.
1594 Returns
1595 -------
1596 None
1598 Raises
1599 ------
1600 NotImplementedError
1601 Raised if file extensions are a concept not understood by this
1602 formatter.
1603 ValueError
1604 Raised if the formatter does not understand this extension.
1606 Notes
1607 -----
1608 This method is available to all Formatters but might not be
1609 implemented by all formatters. It requires that a formatter set
1610 an ``extension`` attribute containing the file extension used when
1611 writing files. If ``extension`` is `None` only the set of supported
1612 extensions will be examined.
1613 """
1614 supported = set(cls.supportedExtensions)
1616 try:
1617 # We are deliberately allowing extension to be undefined by
1618 # default in the base class and mypy complains.
1619 default = cls.extension # type: ignore
1620 except AttributeError:
1621 raise NotImplementedError("No file extension registered with this formatter") from None
1623 # If extension is implemented as an instance property it won't return
1624 # a string when called as a class property. Assume that
1625 # the supported extensions class property is complete.
1626 if default is not None and isinstance(default, str):
1627 supported.add(default)
1629 # Get the file name from the uri
1630 file = location.uri.basename()
1632 # Check that this file name ends with one of the supported extensions.
1633 # This is less prone to confusion than asking the location for
1634 # its extension and then doing a set comparison
1635 for ext in supported:
1636 if file.endswith(ext):
1637 return
1639 raise ValueError(
1640 f"Extension '{location.getExtension()}' on '{location}' "
1641 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})"
1642 )
1644 def predict_path(self) -> str:
1645 """Return the path that would be returned by write.
1647 Does not write any data file.
1649 Uses the `FileDescriptor` associated with the instance.
1651 Returns
1652 -------
1653 path : `str`
1654 Path within datastore that would be associated with the location
1655 stored in this `Formatter`.
1656 """
1657 updated = self.make_updated_location(self.fileDescriptor.location)
1658 return updated.pathInStore.path
1660 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
1661 """Segregate the supplied parameters.
1663 This splits the parameters into those understood by the
1664 formatter and those not understood by the formatter.
1666 Any unsupported parameters are assumed to be usable by associated
1667 assemblers.
1669 Parameters
1670 ----------
1671 parameters : `dict`, optional
1672 Parameters with values that have been supplied by the caller
1673 and which might be relevant for the formatter. If `None`
1674 parameters will be read from the registered `FileDescriptor`.
1676 Returns
1677 -------
1678 supported : `dict`
1679 Those parameters supported by this formatter.
1680 unsupported : `dict`
1681 Those parameters not supported by this formatter.
1682 """
1683 if parameters is None:
1684 parameters = self.fileDescriptor.parameters
1686 if parameters is None:
1687 return {}, {}
1689 if self.unsupportedParameters is None:
1690 # Support none of the parameters
1691 return {}, parameters.copy()
1693 # Start by assuming all are supported
1694 supported = parameters.copy()
1695 unsupported = {}
1697 # And remove any we know are not supported
1698 for p in set(supported):
1699 if p in self.unsupportedParameters:
1700 unsupported[p] = supported.pop(p)
1702 return supported, unsupported
1704 # Support classic V1 interface.
1705 makeUpdatedLocation = make_updated_location
1706 validateExtension = validate_extension
1707 segregateParameters = segregate_parameters
1708 predictPath = predict_path
1710 # Compatibility with V2 properties.
1711 @property
1712 def write_parameters(self) -> Mapping[str, Any]:
1713 return self.writeParameters
1715 @property
1716 def write_recipes(self) -> Mapping[str, Any]:
1717 return self.writeRecipes
1719 @property
1720 def file_descriptor(self) -> FileDescriptor:
1721 return self.fileDescriptor
1723 @property
1724 def data_id(self) -> DataCoordinate:
1725 return self.dataId
1728class FormatterFactory:
1729 """Factory for `Formatter` instances."""
1731 defaultKey = LookupKey("default")
1732 """Configuration key associated with default write parameter settings."""
1734 writeRecipesKey = LookupKey("write_recipes")
1735 """Configuration key associated with write recipes."""
1737 def __init__(self) -> None:
1738 self._mappingFactory = MappingFactory(Formatter)
1740 def __contains__(self, key: LookupKey | str) -> bool:
1741 """Indicate whether the supplied key is present in the factory.
1743 Parameters
1744 ----------
1745 key : `LookupKey`, `str` or objects with ``name`` attribute
1746 Key to use to lookup in the factory whether a corresponding
1747 formatter is present.
1749 Returns
1750 -------
1751 in : `bool`
1752 `True` if the supplied key is present in the factory.
1753 """
1754 return key in self._mappingFactory
1756 def registerFormatters(self, config: Config, *, universe: DimensionUniverse) -> None:
1757 """Bulk register formatters from a config.
1759 Parameters
1760 ----------
1761 config : `Config`
1762 ``formatters`` section of a configuration.
1763 universe : `DimensionUniverse`, optional
1764 Set of all known dimensions, used to expand and validate any used
1765 in lookup keys.
1767 Notes
1768 -----
1769 The configuration can include one level of hierarchy where an
1770 instrument-specific section can be defined to override more general
1771 template specifications. This is represented in YAML using a
1772 key of form ``instrument<name>`` which can then define templates
1773 that will be returned if a `DatasetRef` contains a matching instrument
1774 name in the data ID.
1776 The config is parsed using the function
1777 `~lsst.daf.butler.configSubset.processLookupConfigs`.
1779 The values for formatter entries can be either a simple string
1780 referring to a python type or a dict representing the formatter and
1781 parameters to be hard-coded into the formatter constructor. For
1782 the dict case the following keys are supported:
1784 - formatter: The python type to be used as the formatter class.
1785 - parameters: A further dict to be passed directly to the
1786 ``write_parameters`` Formatter constructor to seed it.
1787 These parameters are validated at instance creation and not at
1788 configuration.
1790 Additionally, a special ``default`` section can be defined that
1791 uses the formatter type (class) name as the keys and specifies
1792 default write parameters that should be used whenever an instance
1793 of that class is constructed.
1795 .. code-block:: yaml
1797 formatters:
1798 default:
1799 lsst.daf.butler.formatters.example.ExampleFormatter:
1800 max: 10
1801 min: 2
1802 comment: Default comment
1803 calexp: lsst.daf.butler.formatters.example.ExampleFormatter
1804 coadd:
1805 formatter: lsst.daf.butler.formatters.example.ExampleFormatter
1806 parameters:
1807 max: 5
1809 Any time an ``ExampleFormatter`` is constructed it will use those
1810 parameters. If an explicit entry later in the configuration specifies
1811 a different set of parameters, the two will be merged with the later
1812 entry taking priority. In the example above ``calexp`` will use
1813 the default parameters but ``coadd`` will override the value for
1814 ``max``.
1816 Formatter configuration can also include a special section describing
1817 collections of write parameters that can be accessed through a
1818 simple label. This allows common collections of options to be
1819 specified in one place in the configuration and reused later.
1820 The ``write_recipes`` section is indexed by Formatter class name
1821 and each key is the label to associate with the parameters.
1823 .. code-block:: yaml
1825 formatters:
1826 write_recipes:
1827 lsst.obs.base.formatters.fitsExposure.FixExposureFormatter:
1828 lossless:
1829 ...
1830 noCompression:
1831 ...
1833 By convention a formatter that uses write recipes will support a
1834 ``recipe`` write parameter that will refer to a recipe name in
1835 the ``write_recipes`` component. The `Formatter` will be constructed
1836 in the `FormatterFactory` with all the relevant recipes and
1837 will not attempt to filter by looking at ``write_parameters`` in
1838 advance. See the specific formatter documentation for details on
1839 acceptable recipe options.
1840 """
1841 allowed_keys = {"formatter", "parameters"}
1843 contents = processLookupConfigs(config, allow_hierarchy=True, universe=universe)
1845 # Extract any default parameter settings
1846 defaultParameters = contents.get(self.defaultKey, {})
1847 if not isinstance(defaultParameters, Mapping):
1848 raise RuntimeError(
1849 "Default formatter parameters in config can not be a single string"
1850 f" (got: {type(defaultParameters)})"
1851 )
1853 # Extract any global write recipes -- these are indexed by
1854 # Formatter class name.
1855 writeRecipes = contents.get(self.writeRecipesKey, {})
1856 if isinstance(writeRecipes, str):
1857 raise RuntimeError(
1858 f"The formatters.{self.writeRecipesKey} section must refer to a dict not '{writeRecipes}'"
1859 )
1861 for key, f in contents.items():
1862 # default is handled in a special way
1863 if key == self.defaultKey:
1864 continue
1865 if key == self.writeRecipesKey:
1866 continue
1868 # Can be a str or a dict.
1869 specificWriteParameters = {}
1870 if isinstance(f, str):
1871 formatter = f
1872 elif isinstance(f, Mapping):
1873 all_keys = set(f)
1874 unexpected_keys = all_keys - allowed_keys
1875 if unexpected_keys:
1876 raise ValueError(f"Formatter {key} uses unexpected keys {unexpected_keys} in config")
1877 if "formatter" not in f:
1878 raise ValueError(f"Mandatory 'formatter' key missing for formatter key {key}")
1879 formatter = f["formatter"]
1880 if "parameters" in f:
1881 specificWriteParameters = f["parameters"]
1882 else:
1883 raise ValueError(f"Formatter for key {key} has unexpected value: '{f}'")
1885 # Apply any default parameters for this formatter
1886 writeParameters = copy.deepcopy(defaultParameters.get(formatter, {}))
1887 writeParameters.update(specificWriteParameters)
1889 kwargs: dict[str, Any] = {}
1890 if writeParameters:
1891 kwargs["write_parameters"] = writeParameters
1893 if formatter in writeRecipes:
1894 kwargs["write_recipes"] = writeRecipes[formatter]
1896 self.registerFormatter(key, formatter, **kwargs)
1898 def getLookupKeys(self) -> set[LookupKey]:
1899 """Retrieve the look up keys for all the registry entries.
1901 Returns
1902 -------
1903 keys : `set` of `LookupKey`
1904 The keys available for matching in the registry.
1905 """
1906 return self._mappingFactory.getLookupKeys()
1908 def getFormatterClassWithMatch(
1909 self, entity: Entity
1910 ) -> tuple[LookupKey, type[Formatter | FormatterV2], dict[str, Any]]:
1911 """Get the matching formatter class along with the registry key.
1913 Parameters
1914 ----------
1915 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1916 Entity to use to determine the formatter to return.
1917 `StorageClass` will be used as a last resort if `DatasetRef`
1918 or `DatasetType` instance is provided. Supports instrument
1919 override if a `DatasetRef` is provided configured with an
1920 ``instrument`` value for the data ID.
1922 Returns
1923 -------
1924 matchKey : `LookupKey`
1925 The key that resulted in the successful match.
1926 formatter : `type`
1927 The class of the registered formatter.
1928 formatter_kwargs : `dict`
1929 Keyword arguments that are associated with this formatter entry.
1930 """
1931 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
1932 matchKey, formatter, formatter_kwargs = self._mappingFactory.getClassFromRegistryWithMatch(names)
1933 log.debug(
1934 "Retrieved formatter %s from key '%s' for entity '%s'",
1935 get_full_type_name(formatter),
1936 matchKey,
1937 entity,
1938 )
1940 return matchKey, formatter, formatter_kwargs
1942 def getFormatterClass(self, entity: Entity) -> type:
1943 """Get the matching formatter class.
1945 Parameters
1946 ----------
1947 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1948 Entity to use to determine the formatter to return.
1949 `StorageClass` will be used as a last resort if `DatasetRef`
1950 or `DatasetType` instance is provided. Supports instrument
1951 override if a `DatasetRef` is provided configured with an
1952 ``instrument`` value for the data ID.
1954 Returns
1955 -------
1956 formatter : `type`
1957 The class of the registered formatter.
1958 """
1959 _, formatter, _ = self.getFormatterClassWithMatch(entity)
1960 return formatter
1962 def getFormatterWithMatch(
1963 self, entity: Entity, *args: Any, **kwargs: Any
1964 ) -> tuple[LookupKey, Formatter | FormatterV2]:
1965 """Get a new formatter instance along with the matching registry key.
1967 Parameters
1968 ----------
1969 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1970 Entity to use to determine the formatter to return.
1971 `StorageClass` will be used as a last resort if `DatasetRef`
1972 or `DatasetType` instance is provided. Supports instrument
1973 override if a `DatasetRef` is provided configured with an
1974 ``instrument`` value for the data ID.
1975 *args : `tuple`
1976 Positional arguments to use pass to the object constructor.
1977 **kwargs
1978 Keyword arguments to pass to object constructor.
1980 Returns
1981 -------
1982 matchKey : `LookupKey`
1983 The key that resulted in the successful match.
1984 formatter : `Formatter`
1985 An instance of the registered formatter.
1986 """
1987 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
1988 matchKey, formatter = self._mappingFactory.getFromRegistryWithMatch(names, *args, **kwargs)
1989 log.debug(
1990 "Retrieved formatter %s from key '%s' for entity '%s'",
1991 get_full_type_name(formatter),
1992 matchKey,
1993 entity,
1994 )
1996 return matchKey, formatter
1998 def getFormatter(self, entity: Entity, *args: Any, **kwargs: Any) -> Formatter | FormatterV2:
1999 """Get a new formatter instance.
2001 Parameters
2002 ----------
2003 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
2004 Entity to use to determine the formatter to return.
2005 `StorageClass` will be used as a last resort if `DatasetRef`
2006 or `DatasetType` instance is provided. Supports instrument
2007 override if a `DatasetRef` is provided configured with an
2008 ``instrument`` value for the data ID.
2009 *args : `tuple`
2010 Positional arguments to use pass to the object constructor.
2011 **kwargs
2012 Keyword arguments to pass to object constructor.
2014 Returns
2015 -------
2016 formatter : `Formatter`
2017 An instance of the registered formatter.
2018 """
2019 _, formatter = self.getFormatterWithMatch(entity, *args, **kwargs)
2020 return formatter
2022 def registerFormatter(
2023 self,
2024 type_: LookupKey | str | StorageClass | DatasetType,
2025 formatter: str,
2026 *,
2027 overwrite: bool = False,
2028 **kwargs: Any,
2029 ) -> None:
2030 """Register a `Formatter`.
2032 Parameters
2033 ----------
2034 type_ : `LookupKey`, `str`, `StorageClass` or `DatasetType`
2035 Type for which this formatter is to be used. If a `LookupKey`
2036 is not provided, one will be constructed from the supplied string
2037 or by using the ``name`` property of the supplied entity.
2038 formatter : `str` or class of type `Formatter`
2039 Identifies a `Formatter` subclass to use for reading and writing
2040 Datasets of this type. Can be a `Formatter` class.
2041 overwrite : `bool`, optional
2042 If `True` an existing entry will be replaced by the new value.
2043 Default is `False`.
2044 **kwargs
2045 Keyword arguments to always pass to object constructor when
2046 retrieved.
2048 Raises
2049 ------
2050 ValueError
2051 Raised if the formatter does not name a valid formatter type and
2052 ``overwrite`` is `False`.
2053 """
2054 self._mappingFactory.placeInRegistry(type_, formatter, overwrite=overwrite, **kwargs)
2057class FormatterV1inV2(FormatterV2):
2058 """An implementation of a V2 formatter that provides a compatibility
2059 interface for V1 formatters.
2061 Parameters
2062 ----------
2063 file_descriptor : `FileDescriptor`, optional
2064 Identifies the file to read or write, and the associated storage
2065 classes and parameter information. Its value can be `None` if the
2066 caller will never call `Formatter.read` or `Formatter.write`.
2067 ref : `DatasetRef`
2068 The dataset associated with this formatter. Should not be a component
2069 dataset ref.
2070 formatter : `Formatter`
2071 A version 1 `Formatter` instance. The V2 formatter layer forwards calls
2072 to this formatter.
2073 write_parameters : `dict`, optional
2074 Any parameters to be hard-coded into this instance to control how
2075 the dataset is serialized.
2076 write_recipes : `dict`, optional
2077 Detailed write Recipes indexed by recipe name.
2078 **kwargs
2079 Additional arguments that will be ignored but allow for
2080 `Formatter` V1 parameters to be given.
2081 """
2083 can_read_from_local_file = True
2084 """This formatter can read from a local file."""
2086 def __init__(
2087 self,
2088 file_descriptor: FileDescriptor,
2089 *,
2090 ref: DatasetRef,
2091 formatter: Formatter,
2092 write_parameters: Mapping[str, Any] | None = None,
2093 write_recipes: Mapping[str, Any] | None = None,
2094 # Compatibility parameters. Unused in v2.
2095 **kwargs: Any,
2096 ):
2097 if not isinstance(formatter, Formatter):
2098 raise TypeError(f"Formatter parameter was not a V1 formatter (was {type(formatter)})")
2100 # Replace the class property with instance values from this
2101 # V1 formatter so that the V2 __init__ will be able to validate.
2102 self.supported_write_parameters = formatter.supportedWriteParameters # type: ignore
2103 self._formatter = formatter
2105 super().__init__(
2106 file_descriptor, ref=ref, write_parameters=write_parameters, write_recipes=write_recipes
2107 )
2109 def get_write_extension(self) -> str:
2110 ext = self._formatter.extension
2111 return ext if ext is not None else ""
2113 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
2114 return self._formatter.segregate_parameters(parameters)
2116 def validate_write_recipes( # type: ignore
2117 self,
2118 recipes: Mapping[str, Any] | None,
2119 ) -> Mapping[str, Any] | None:
2120 # This should be a class method but a class method can not work
2121 # for a dynamic shim. Luckily the shim is only used as an instance.
2122 return self._formatter.validate_write_recipes(recipes)
2124 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any:
2125 # Need to temporarily override the location since the V1 formatter
2126 # will not know anything about this local file.
2128 # V2 does not have a fromBytes equivalent.
2129 if self._formatter.can_read_bytes():
2130 with open(path, "rb") as fd:
2131 serialized_dataset = fd.read()
2132 return self._formatter.fromBytes(serialized_dataset, component=component)
2134 location = Location(None, path)
2135 with self._formatter._updateLocation(location):
2136 try:
2137 result = self._formatter.read(component=component)
2138 except NotImplementedError:
2139 # V1 raises NotImplementedError but V2 is expecting something
2140 # slightly different.
2141 return NotImplemented
2142 return result
2144 def to_bytes(self, in_memory_dataset: Any) -> bytes:
2145 try:
2146 return self._formatter.toBytes(in_memory_dataset)
2147 except NotImplementedError as e:
2148 # V1 raises NotImplementedError but V2 is expecting something
2149 # slightly different.
2150 raise FormatterNotImplementedError(str(e)) from e
2152 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
2153 with self._formatter._updateLocation(Location(None, uri)):
2154 try:
2155 self._formatter.write(in_memory_dataset)
2156 except NotImplementedError as e:
2157 # V1 raises NotImplementedError but V2 is expecting something
2158 # slightly different.
2159 raise FormatterNotImplementedError(str(e)) from e
2162# Type to use when allowing a Formatter or its class name
2163FormatterParameter: TypeAlias = str | type[Formatter] | Formatter | FormatterV2 | type[FormatterV2]