Coverage for python/lsst/daf/butler/_formatter.py: 23%
531 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-07 01:48 -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 .dimensions import DataCoordinate, DimensionUniverse
58from .mapping_factory import MappingFactory
60log = logging.getLogger(__name__)
62if TYPE_CHECKING:
63 from ._dataset_provenance import DatasetProvenance
64 from ._dataset_ref import DatasetRef
65 from ._dataset_type import DatasetType
66 from ._storage_class import StorageClass
67 from .datastore.cache_manager import AbstractDatastoreCacheManager
69 # Define a new special type for functions that take "entity"
70 Entity: TypeAlias = DatasetType | DatasetRef | StorageClass | str
73class FileIntegrityError(RuntimeError):
74 """The file metadata is inconsistent with the metadata supplied by
75 the datastore.
76 """
79class FormatterNotImplementedError(NotImplementedError):
80 """Formatter does not implement the specific read or write method
81 that is being requested.
82 """
85class FormatterV2:
86 """Interface for reading and writing datasets using URIs.
88 The formatters are associated with a particular `StorageClass`.
90 Parameters
91 ----------
92 file_descriptor : `FileDescriptor`, optional
93 Identifies the file to read or write, and the associated storage
94 classes and parameter information.
95 ref : `DatasetRef`
96 The dataset associated with this formatter. Should not be a component
97 dataset ref.
98 write_parameters : `dict`, optional
99 Parameters to control how the dataset is serialized.
100 write_recipes : `dict`, optional
101 Detailed write recipes indexed by recipe name.
103 **kwargs
104 Additional arguments that will be ignored but allow for
105 `Formatter` V1 parameters to be given.
107 Notes
108 -----
109 A `FormatterV2` author should not override the default `read` or `write`
110 method. Instead for read the formatter author should implement one or all
111 of `read_from_stream`, `read_from_uri`, or `read_from_local_file`. The
112 method `read_from_uri` will always be attempted first and could be more
113 efficient (since it allows the possibility for a subset of the data file to
114 be accessed remotely when parameters or components are specified) but it
115 will not update the local cache. If the entire contents of the remote file
116 are being accessed (no component or parameters defined) and the dataset
117 would be cached, `read_from_uri` will be called with a local file. If the
118 file is remote and the parameters that have been included are known to be
119 more efficiently handled with a local file, the `read_from_uri` method can
120 return `NotImplemented` to indicate that a local file should be given
121 instead.
123 Similarly for writes, the `write` method can not be subclassed. Instead
124 the formatter author should implement `to_bytes` or `write_local_file`.
125 For local URIs the system will always call `write_local_file` first (which
126 by default will call `to_bytes`) to ensure atomic writes are implemented.
127 For remote URIs with local caching disabled, `to_bytes` will be called
128 first and the remote updated directly. If the dataset should be cached
129 it will always be written locally first.
130 """
132 unsupported_parameters: ClassVar[Set[str] | None] = frozenset()
133 """Set of read parameters not understood by this `Formatter`. An empty set
134 means all parameters are supported. `None` indicates that no parameters
135 are supported. These parameters should match those defined in the storage
136 class definition. (`frozenset`).
137 """
139 supported_write_parameters: ClassVar[Set[str] | None] = None
140 """Parameters understood by this formatter that can be used to control
141 how a dataset is serialized. `None` indicates that no parameters are
142 supported."""
144 default_extension: ClassVar[str | None] = None
145 """Default extension to use when writing a file.
147 Can be `None` if the extension is determined dynamically. Use the
148 `get_write_extension` method to get the actual extension to use.
149 """
151 supported_extensions: ClassVar[Set[str]] = frozenset()
152 """Set of all extensions supported by this formatter.
154 Any extension assigned to the ``default_extension`` property will be
155 automatically included in the list of supported extensions.
156 """
158 can_read_from_uri: ClassVar[bool] = False
159 """Declare whether `read_from_uri` is available to this formatter."""
161 can_read_from_stream: ClassVar[bool] = False
162 """Declare whether `read_from_stream` is available to this formatter."""
164 can_read_from_local_file: ClassVar[bool] = False
165 """Declare whether `read_from_file` is available to this formatter."""
167 def __init__(
168 self,
169 file_descriptor: FileDescriptor,
170 *,
171 ref: DatasetRef,
172 write_parameters: Mapping[str, Any] | None = None,
173 write_recipes: Mapping[str, Any] | None = None,
174 # Compatibility parameters. Unused in v2.
175 **kwargs: Any,
176 ):
177 if not isinstance(file_descriptor, FileDescriptor):
178 raise TypeError("File descriptor must be a FileDescriptor")
180 self._file_descriptor = file_descriptor
182 if ref.isComponent():
183 # It is a component ref for disassembled composites.
184 ref = ref.makeCompositeRef()
185 self._dataset_ref = ref
187 # Check that the write parameters are allowed
188 if write_parameters:
189 if self.supported_write_parameters is None:
190 raise ValueError(
191 f"This formatter does not accept any write parameters. Got: {', '.join(write_parameters)}"
192 )
193 else:
194 given = set(write_parameters)
195 unknown = given - self.supported_write_parameters
196 if unknown:
197 s = "s" if len(unknown) != 1 else ""
198 unknownStr = ", ".join(f"'{u}'" for u in unknown)
199 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}")
201 self._write_parameters = write_parameters
202 self._write_recipes = self.validate_write_recipes(write_recipes)
204 def __str__(self) -> str:
205 return f"{self.name()}@{self.file_descriptor.location.uri}"
207 def __repr__(self) -> str:
208 return f"{self.name()}({self.file_descriptor!r})"
210 @property
211 def file_descriptor(self) -> FileDescriptor:
212 """File descriptor associated with this formatter
213 (`FileDescriptor`).
214 """
215 return self._file_descriptor
217 @property
218 def data_id(self) -> DataCoordinate:
219 """Return Data ID associated with this formatter (`DataCoordinate`)."""
220 return self._dataset_ref.dataId
222 @property
223 def dataset_ref(self) -> DatasetRef:
224 """Return Dataset Ref associated with this formatter (`DatasetRef`)."""
225 return self._dataset_ref
227 @property
228 def write_parameters(self) -> Mapping[str, Any]:
229 """Parameters to use when writing out datasets."""
230 if self._write_parameters is not None:
231 return self._write_parameters
232 return {}
234 @property
235 def write_recipes(self) -> Mapping[str, Any]:
236 """Detailed write Recipes indexed by recipe name."""
237 if self._write_recipes is not None:
238 return self._write_recipes
239 return {}
241 def get_write_extension(self) -> str:
242 """Extension to use when writing a file."""
243 default_extension = self.default_extension
244 extension = default_extension if default_extension is not None else ""
245 return extension
247 def can_accept(self, in_memory_dataset: Any) -> bool:
248 """Indicate whether this formatter can accept the specified
249 storage class directly.
251 Parameters
252 ----------
253 in_memory_dataset : `object`
254 The dataset that is to be written.
256 Returns
257 -------
258 accepts : `bool`
259 If `True` the formatter can write data of this type without
260 requiring datastore to convert it. If `False` the datastore
261 will attempt to convert before writing.
263 Notes
264 -----
265 The base class always returns `False` even if the given type is an
266 instance of the storage class type. This will result in a storage
267 class conversion no-op but also allows mocks with mocked storage
268 classes to work properly.
269 """
270 return False
272 @classmethod
273 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
274 """Validate supplied recipes for this formatter.
276 The recipes are supplemented with default values where appropriate.
278 Parameters
279 ----------
280 recipes : `dict`
281 Recipes to validate.
283 Returns
284 -------
285 validated : `dict`
286 Validated recipes.
288 Raises
289 ------
290 RuntimeError
291 Raised if validation fails. The default implementation raises
292 if any recipes are given.
293 """
294 if recipes:
295 raise RuntimeError(f"This formatter does not understand these write recipes: {recipes}")
296 return recipes
298 @classmethod
299 def name(cls) -> str:
300 """Return the fully qualified name of the formatter.
302 Returns
303 -------
304 name : `str`
305 Fully-qualified name of formatter class.
306 """
307 return get_full_type_name(cls)
309 def _is_disassembled(self) -> bool:
310 """Return `True` if this formatter is looking at a disassembled
311 component.
312 """
313 return self.file_descriptor.component is not None
315 def _check_resource_size(self, uri: ResourcePath, recorded_size: int, resource_size: int) -> None:
316 """Compare the recorded size with the resource size.
318 The given URI will not be accessed.
319 """
320 if recorded_size >= 0 and resource_size != recorded_size:
321 raise FileIntegrityError(
322 "Integrity failure in Datastore. "
323 f"Size of file {uri} ({resource_size}) "
324 f"does not match size recorded in registry of {recorded_size}"
325 )
327 def _get_cache_ref(self) -> DatasetRef:
328 """Get the `DatasetRef` to use for cache look ups.
330 Returns
331 -------
332 ref : `lsst.daf.butler.DatasetRef`
333 The dataset ref to use when looking in the cache.
334 For single-file dataset this will be the dataset ref directly.
335 If this is disassembled we need the component and the component
336 will be in the `FileDescriptor`.
337 """
338 if self.file_descriptor.component is None:
339 cache_ref = self.dataset_ref
340 else:
341 cache_ref = self.dataset_ref.makeComponentRef(self.file_descriptor.component)
342 return cache_ref
344 def _ensure_cache(
345 self, cache_manager: AbstractDatastoreCacheManager | None = None
346 ) -> AbstractDatastoreCacheManager:
347 """Return the cache if given else return a null cache."""
348 if cache_manager is None:
349 # Circular import avoidance.
350 from .datastore.cache_manager import DatastoreDisabledCacheManager
352 cache_manager = DatastoreDisabledCacheManager(None, None)
353 return cache_manager
355 def read(
356 self,
357 component: str | None = None,
358 expected_size: int = -1,
359 cache_manager: AbstractDatastoreCacheManager | None = None,
360 ) -> Any:
361 """Read a Dataset.
363 Parameters
364 ----------
365 component : `str`, optional
366 Component to read from the file. Only used if the `StorageClass`
367 for reading differed from the `StorageClass` used to write the
368 file.
369 expected_size : `int`, optional
370 If known, the expected size of the resource to read. This can be
371 used for verification or to decide whether to do a direct read or a
372 file download. ``-1`` indicates the file size is not known.
373 cache_manager : `AbstractDatastoreCacheManager`
374 A cache manager to use to allow a formatter to cache a remote file
375 locally or read a cached file that is already local.
377 Returns
378 -------
379 in_memory_dataset : `object`
380 The requested Dataset.
382 Notes
383 -----
384 This method should not be subclassed. Instead formatter subclasses
385 should re-implement the specific `read_from_*` methods as appropriate.
386 Each of these methods has a corresponding class property that must
387 be `True` for the method to be called.
389 The priority for reading is:
391 * `read_from_uri`
392 * `read_from_stream`
393 * `read_from_local_file`
394 * `read_from_uri` (but with a local file)
396 Any of these methods can return `NotImplemented` if there is a desire
397 to skip to the next one in the list. If a dataset is being requested
398 with no component, no parameters, and it should also be added to the
399 local cache, the first two calls will be skipped (unless
400 `read_from_stream` is the only implemented read method) such that a
401 local file will be used.
403 A Formatter can also read a file from within a Zip file if the
404 URI associated with the `FileDescriptor` corresponds to a file with
405 a `.zip` extension and a URI fragment of the form
406 ``zip-path={path_in_zip}``. When reading a file from within a Zip
407 file the priority for reading is:
409 * `read_from_stream`
410 * `read_from_local_file`
411 * `read_from_uri`
413 There are multiple cases that must be handled for reading:
415 For a single file:
417 * No component requested, read the whole file.
418 * Component requested, optionally read the component efficiently,
419 else read the whole file and extract the component.
420 * Derived component requested, read whole file or read relevant
421 component and derive.
423 Disassembled Composite:
425 * The file to read here is the component itself. Formatter only knows
426 about this one component file. Should be no component specified
427 in the ``read`` call but the `FileDescriptor` will know which
428 component this is.
429 * A derived component. The file to read is a component but not the
430 specified component. The caching needs the component from which
431 it's derived.
433 Raises
434 ------
435 FormatterNotImplementedError
436 Raised if no implementations were found that could read this
437 resource.
438 """
439 # If the file to read is a ZIP file with a fragment requesting
440 # a file within the ZIP file, it is no longer possible to use the
441 # direct read from URI option and the contents of the Zip file must
442 # be extracted.
443 uri = self.file_descriptor.location.uri
444 if uri.fragment and uri.unquoted_fragment.startswith("zip-path="):
445 _, _, path_in_zip = uri.unquoted_fragment.partition("=")
447 # Open the Zip file using ResourcePath.
448 with uri.open("rb") as fd:
449 with zipfile.ZipFile(fd) as zf: # type: ignore
450 if self.can_read_from_stream:
451 with contextlib.closing(zf.open(path_in_zip)) as zip_fd:
452 result = self.read_from_stream(zip_fd, component, expected_size=expected_size)
454 if result is not NotImplemented:
455 return result
457 # For now for both URI and local file options we retrieve
458 # the bytes to a temporary local and use that.
459 _, suffix = os.path.splitext(path_in_zip)
460 with ResourcePath.temporary_uri(suffix=suffix) as tmp_uri:
461 tmp_uri.write(zf.read(path_in_zip))
463 if self.can_read_from_local_file:
464 result = self.read_from_local_file(
465 tmp_uri.ospath, component, expected_size=expected_size
466 )
467 if result is not NotImplemented:
468 return result
469 if self.can_read_from_uri:
470 result = self.read_from_uri(tmp_uri, component, expected_size=expected_size)
471 if result is not NotImplemented:
472 return result
474 raise FormatterNotImplementedError(
475 f"Formatter {self.name()} could not read the file at {uri} using any method."
476 )
478 # If the there are no parameters, no component request, and
479 # the file should be cached, it is better to defer the read_from_uri
480 # to use the local file and populate the cache.
481 prefer_local = False
482 if (
483 component is None
484 and not self.file_descriptor.parameters
485 and self._ensure_cache(cache_manager).should_be_cached(self._get_cache_ref())
486 ):
487 prefer_local = True
489 # First see if the formatter can support direct remote read from
490 # a URI. This can be called later from the local path. If it returns
491 # NotImplemented the formatter decided on its own that local
492 # reads are preferred.
493 if not prefer_local and self.can_read_from_uri:
494 result = self.read_directly_from_possibly_cached_uri(
495 component, expected_size, cache_manager=cache_manager
496 )
497 if result is not NotImplemented:
498 return result
500 # Some formatters might want to be able to read directly from
501 # an open file stream. This is preferred over forcing a download
502 # to local file system unless a file read option is available and we
503 # want to store it in the cache because the whole file is being read.
504 if self.can_read_from_stream and not (
505 prefer_local and (self.can_read_from_uri or self.can_read_from_local_file)
506 ):
507 result = self.read_from_possibly_cached_stream(
508 component, expected_size, cache_manager=cache_manager
509 )
510 if result is not NotImplemented:
511 return result
513 # Finally, try to read the local file.
514 if self.can_read_from_local_file or self.can_read_from_uri:
515 result = self.read_from_possibly_cached_local_file(
516 component, expected_size, cache_manager=cache_manager
517 )
518 if result is not NotImplemented:
519 return result
521 raise FormatterNotImplementedError(
522 f"Formatter {self.name()} could not read the file at {uri} using any method."
523 )
525 def _read_from_possibly_cached_location_no_cache_write(
526 self,
527 callback: Callable[[ResourcePath, str | None, int], Any],
528 component: str | None = None,
529 expected_size: int = -1,
530 *,
531 cache_manager: AbstractDatastoreCacheManager | None = None,
532 ) -> Any:
533 """Read from the cache and call payload without writing to cache."""
534 cache_manager = self._ensure_cache(cache_manager)
536 uri = self.file_descriptor.location.uri
537 cache_ref = self._get_cache_ref()
539 # The component for log messages is either the component requested
540 # explicitly or the component from the file descriptor.
541 log_component = component if component is not None else self.file_descriptor.component
543 with cache_manager.find_in_cache(cache_ref, uri.getExtension()) as cached_file:
544 if cached_file is not None:
545 desired_uri = cached_file
546 msg = f" (cached version of {uri})"
547 else:
548 desired_uri = uri
549 msg = ""
551 if desired_uri.isLocal:
552 # Do not spend the time doing a slow size() call to a remote
553 # resource.
554 self._check_resource_size(desired_uri, expected_size, desired_uri.size())
556 with time_this(
557 log,
558 msg="Reading%s from file handle %s%s with formatter %s",
559 args=(
560 f" component {log_component}" if log_component else "",
561 desired_uri,
562 msg,
563 self.name(),
564 ),
565 ):
566 return callback(desired_uri, component, expected_size)
568 def read_from_possibly_cached_stream(
569 self,
570 component: str | None = None,
571 expected_size: int = -1,
572 *,
573 cache_manager: AbstractDatastoreCacheManager | None = None,
574 ) -> Any:
575 """Read from a stream, checking for possible presence in local cache.
577 Parameters
578 ----------
579 component : `str`, optional
580 Component to read from the file. Only used if the `StorageClass`
581 for reading differed from the `StorageClass` used to write the
582 file.
583 expected_size : `int`, optional
584 If known, the expected size of the resource to read. This can be
585 used for verification or to decide whether to do a direct read or a
586 file download. ``-1`` indicates the file size is not known.
587 cache_manager : `AbstractDatastoreCacheManager`
588 A cache manager to use to allow a formatter to check if there is
589 a copy of the file in the local cache.
591 Returns
592 -------
593 in_memory_dataset : `object` or `NotImplemented`
594 The requested Dataset or an indication that the read mode was
595 not implemented.
597 Notes
598 -----
599 Calls `read_from_stream` but will first check the datastore cache
600 in case the file is present locally. This method will not download
601 a file to the local cache.
602 """
604 def _open_stream(uri: ResourcePath, comp: str | None, size: int = -1) -> Any:
605 with uri.open("rb") as fd:
606 return self.read_from_stream(fd, comp, expected_size=size)
608 return self._read_from_possibly_cached_location_no_cache_write(
609 _open_stream, component, expected_size=expected_size, cache_manager=cache_manager
610 )
612 def read_directly_from_possibly_cached_uri(
613 self,
614 component: str | None = None,
615 expected_size: int = -1,
616 *,
617 cache_manager: AbstractDatastoreCacheManager | None = None,
618 ) -> Any:
619 """Read from arbitrary URI, checking for possible presence in local
620 cache.
622 Parameters
623 ----------
624 component : `str`, optional
625 Component to read from the file. Only used if the `StorageClass`
626 for reading differed from the `StorageClass` used to write the
627 file.
628 expected_size : `int`, optional
629 If known, the expected size of the resource to read. This can be
630 ``-1`` indicates the file size is not known.
631 cache_manager : `AbstractDatastoreCacheManager`
632 A cache manager to use to allow a formatter to check if there is
633 a copy of the file in the local cache.
635 Returns
636 -------
637 in_memory_dataset : `object` or `NotImplemented`
638 The requested Dataset or an indication that the read mode was
639 not implemented.
641 Notes
642 -----
643 This method will first check the datastore cache
644 in case the file is present locally. This method will not cache a
645 remote dataset and will only do a size check for local files to avoid
646 unnecessary round trips to a remote server.
648 The URI will be read by calling `read_from_uri`.
649 """
651 def _open_uri(uri: ResourcePath, comp: str | None, size: int = -1) -> Any:
652 return self.read_from_uri(uri, comp, expected_size=size)
654 return self._read_from_possibly_cached_location_no_cache_write(
655 _open_uri, component, expected_size=expected_size, cache_manager=cache_manager
656 )
658 def read_from_possibly_cached_local_file(
659 self,
660 component: str | None = None,
661 expected_size: int = -1,
662 *,
663 cache_manager: AbstractDatastoreCacheManager | None = None,
664 ) -> Any:
665 """Read a dataset ensuring that a local file is used, checking the
666 cache for it.
668 Parameters
669 ----------
670 component : `str`, optional
671 Component to read from the file. Only used if the `StorageClass`
672 for reading differed from the `StorageClass` used to write the
673 file.
674 expected_size : `int`, optional
675 If known, the expected size of the resource to read. This can be
676 used for verification or to decide whether to do a direct read or a
677 file download. ``-1`` indicates the file size is not known.
678 cache_manager : `AbstractDatastoreCacheManager`
679 A cache manager to use to allow a formatter to cache a remote file
680 locally or read a cached file that is already local.
682 Returns
683 -------
684 in_memory_dataset : `object` or `NotImplemented`
685 The requested Dataset or an indication that the read mode was
686 not implemented.
688 Notes
689 -----
690 The file will be downloaded and cached if it is a remote resource.
691 The file contents will be read using `read_from_local_file` or
692 `read_from_uri`, with preference given to the former.
693 """
694 cache_manager = self._ensure_cache(cache_manager)
695 uri = self.file_descriptor.location.uri
697 # Need to have something we can look up in the cache.
698 cache_ref = self._get_cache_ref()
700 # The component for log messages is either the component requested
701 # explicitly or the component from the file descriptor.
702 log_component = component if component is not None else self.file_descriptor.component
704 result = NotImplemented
706 # Ensure we have a local file.
707 with cache_manager.find_in_cache(cache_ref, uri.getExtension()) as cached_file:
708 if cached_file is not None:
709 msg = f"(via cache read of remote file {uri})"
710 uri = cached_file
711 else:
712 msg = ""
714 with uri.as_local() as local_uri:
715 self._check_resource_size(self.file_descriptor.location.uri, expected_size, local_uri.size())
716 can_be_cached = False
717 if uri != local_uri:
718 # URI was remote and file was downloaded
719 cache_msg = ""
721 if cache_manager.should_be_cached(cache_ref):
722 # In this scenario we want to ask if the downloaded
723 # file should be cached but we should not cache
724 # it until after we've used it (to ensure it can't
725 # be expired whilst we are using it).
726 can_be_cached = True
728 # Say that it is "likely" to be cached because
729 # if the formatter read fails we will not be
730 # caching this file.
731 cache_msg = " and likely cached"
733 msg = f"(via download to local file{cache_msg})"
735 with time_this(
736 log,
737 msg="Reading%s from location %s %s with formatter %s",
738 args=(
739 f" component {log_component}" if log_component else "",
740 uri,
741 msg,
742 self.name(),
743 ),
744 ):
745 if self.can_read_from_local_file:
746 result = self.read_from_local_file(
747 local_uri.ospath, component=component, expected_size=expected_size
748 )
749 if result is NotImplemented and self.can_read_from_uri:
750 # If the direct URI reader was skipped earlier and
751 # there is no explicit local file implementation, pass
752 # in the guaranteed local URI to the generic reader.
753 result = self.read_from_uri(
754 local_uri, component=component, expected_size=expected_size
755 )
757 # File was read successfully so can move to cache.
758 # Also move to cache even if NotImplemented was returned.
759 if can_be_cached:
760 cache_manager.move_to_cache(local_uri, cache_ref)
762 return result
764 def read_from_uri(self, uri: ResourcePath, component: str | None = None, expected_size: int = -1) -> Any:
765 """Read a dataset from a URI that can be local or remote.
767 Parameters
768 ----------
769 uri : `lsst.resources.ResourcePath`
770 URI to use to read the dataset. This URI can be local or remote
771 and can refer to the actual resource or to a locally cached file.
772 component : `str` or `None`, optional
773 The component to be read from the dataset.
774 expected_size : `int`, optional
775 If known, the expected size of the resource to read. This can be
776 ``-1`` indicates the file size is not known.
778 Returns
779 -------
780 in_memory_dataset : `object` or `NotImplemented`
781 The Python object read from the resource or `NotImplemented`.
783 Raises
784 ------
785 FormatterNotImplementedError
786 Raised if there is no support for direct reads from a, possibly,
787 remote URI.
789 Notes
790 -----
791 This method is only called if the class property
792 ``can_read_from_uri`` is set to `True`.
794 It is possible that a cached local file will be given to this method
795 even if it was originally a remote URI. This can happen if the original
796 write resulted in the file being added to the local cache.
798 If the full file is being read this file will not be added to the
799 local cache. Consider returning `NotImplemented` in
800 this situation, for example if there are no parameters or component
801 specified, and allowing the system to fall back to calling
802 `read_from_local_file` (which will populate the cache if configured
803 to do so).
804 """
805 return NotImplemented
807 def read_from_stream(
808 self, stream: BinaryIO | ResourceHandleProtocol, component: str | None = None, expected_size: int = -1
809 ) -> Any:
810 """Read from an open file descriptor.
812 Parameters
813 ----------
814 stream : `lsst.resources.ResourceHandleProtocol` or \
815 `typing.BinaryIO`
816 File stream to use to read the dataset.
817 component : `str` or `None`, optional
818 The component to be read from the dataset.
819 expected_size : `int`, optional
820 If known, the expected size of the resource to read. This can be
821 ``-1`` indicates the file size is not known.
823 Returns
824 -------
825 in_memory_dataset : `object` or `NotImplemented`
826 The Python object read from the stream or `NotImplemented`.
828 Notes
829 -----
830 Only called if the class property ``can_read_from_stream`` is `True`.
831 """
832 return NotImplemented
834 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any:
835 """Read a dataset from a URI guaranteed to refer to the local file
836 system.
838 Parameters
839 ----------
840 path : `str`
841 Path to a local file that should be read.
842 component : `str` or `None`, optional
843 The component to be read from the dataset.
844 expected_size : `int`, optional
845 If known, the expected size of the resource to read. This can be
846 ``-1`` indicates the file size is not known.
848 Returns
849 -------
850 in_memory_dataset : `object` or `NotImplemented`
851 The Python object read from the resource or `NotImplemented`.
853 Raises
854 ------
855 FormatterNotImplementedError
856 Raised if there is no implementation written to read data
857 from a local file.
859 Notes
860 -----
861 This method will only be called if the class property
862 ``can_read_from_local_file`` is `True` and other options were not
863 used.
864 """
865 return NotImplemented
867 def add_provenance(
868 self, in_memory_dataset: Any, /, *, provenance: DatasetProvenance | None = None
869 ) -> Any:
870 """Add provenance to the dataset.
872 Parameters
873 ----------
874 in_memory_dataset : `object`
875 The dataset to serialize.
876 provenance : `DatasetProvenance` or `None`, optional
877 Provenance to attach to dataset.
879 Returns
880 -------
881 dataset_to_write : `object`
882 The dataset to use for serialization. Can be the same object as
883 given.
885 Notes
886 -----
887 The base class implementation returns the given object unchanged.
888 """
889 return in_memory_dataset
891 @final
892 def write(
893 self,
894 in_memory_dataset: Any,
895 /,
896 *,
897 cache_manager: AbstractDatastoreCacheManager | None = None,
898 provenance: DatasetProvenance | None = None,
899 ) -> None:
900 """Write a Dataset.
902 Parameters
903 ----------
904 in_memory_dataset : `object`
905 The Dataset to serialize.
906 cache_manager : `AbstractDatastoreCacheManager`
907 A cache manager to use to allow a formatter to cache the written
908 file.
909 provenance : `DatasetProvenance` | `None`, optional
910 Provenance to attach to the file being written.
912 Raises
913 ------
914 FormatterNotImplementedError
915 Raised if the formatter subclass has not implemented
916 `write_local_file` and `to_bytes` was not called.
917 Exception
918 Raised if there is an error serializing the dataset to disk.
920 Notes
921 -----
922 The intent is for subclasses to implement either `to_bytes` or
923 `write_local_file` or both and not to subclass this method.
924 """
925 # Ensure we are using the correct file extension.
926 uri = self.file_descriptor.location.uri.updatedExtension(self.get_write_extension())
928 # Attach any provenance to the dataset. This could involve returning
929 # a different object.
930 in_memory_dataset = self.add_provenance(in_memory_dataset, provenance=provenance)
932 written = self.write_direct(in_memory_dataset, uri, cache_manager)
933 if not written:
934 self.write_locally_then_move(in_memory_dataset, uri, cache_manager)
936 def write_direct(
937 self,
938 in_memory_dataset: Any,
939 uri: ResourcePath,
940 cache_manager: AbstractDatastoreCacheManager | None = None,
941 ) -> bool:
942 """Serialize and write directly to final location.
944 Parameters
945 ----------
946 in_memory_dataset : `object`
947 The Dataset to serialize.
948 uri : `lsst.resources.ResourcePath`
949 URI to use when writing the serialized dataset.
950 cache_manager : `AbstractDatastoreCacheManager`
951 A cache manager to use to allow a formatter to cache the written
952 file.
954 Returns
955 -------
956 written : `bool`
957 Flag to indicate whether the direct write did happen.
959 Raises
960 ------
961 Exception
962 Raised if there was a failure from serializing to bytes that
963 was not `FormatterNotImplementedError`.
965 Notes
966 -----
967 This method will call `to_bytes` to serialize the in-memory dataset
968 and then will call the `~lsst.resources.ResourcePath.write` method
969 directly.
971 If the dataset should be cached or is local the file will not be
972 written and the method will return `False`. This is because local URIs
973 should be written to a temporary file name and then renamed to allow
974 atomic writes. That path is handled by `write_locally_then_move`
975 through `write_local_file`) and is preferred over this method being
976 subclassed and the atomic write re-implemented.
977 """
978 cache_manager = self._ensure_cache(cache_manager)
980 # For remote URIs some datasets can be serialized directly
981 # to bytes and sent to the remote datastore without writing a
982 # file. If the dataset is intended to be saved to the cache
983 # a file is always written and direct write to the remote
984 # datastore is bypassed.
985 data_written = False
986 if not uri.isLocal and not cache_manager.should_be_cached(self._get_cache_ref()):
987 # Remote URI that is not cached so can write directly.
988 try:
989 serialized_dataset = self.to_bytes(in_memory_dataset)
990 except FormatterNotImplementedError:
991 # Fallback to the file writing option.
992 pass
993 except Exception as e:
994 e.add_note(
995 f"Failed to serialize dataset {self.dataset_ref} of "
996 f"type {get_full_type_name(in_memory_dataset)} to bytes."
997 )
998 raise
999 else:
1000 log.debug("Writing bytes directly to %s", uri)
1001 uri.write(serialized_dataset, overwrite=True)
1002 log.debug("Successfully wrote bytes directly to %s", uri)
1003 data_written = True
1004 return data_written
1006 def write_locally_then_move(
1007 self,
1008 in_memory_dataset: Any,
1009 uri: ResourcePath,
1010 cache_manager: AbstractDatastoreCacheManager | None = None,
1011 ) -> None:
1012 """Write file to file system and then move to final location.
1014 Parameters
1015 ----------
1016 in_memory_dataset : `object`
1017 The Dataset to serialize.
1018 uri : `lsst.resources.ResourcePath`
1019 URI to use when writing the serialized dataset.
1020 cache_manager : `AbstractDatastoreCacheManager`
1021 A cache manager to use to allow a formatter to cache the written
1022 file.
1024 Raises
1025 ------
1026 FormatterNotImplementedError
1027 Raised if the formatter subclass has not implemented
1028 `write_local_file`.
1029 Exception
1030 Raised if there is an error serializing the dataset to disk.
1031 """
1032 cache_manager = self._ensure_cache(cache_manager)
1034 # Always write to a temporary even if
1035 # using a local file system -- that gives us atomic writes.
1036 # If a process is killed as the file is being written we do not
1037 # want it to remain in the correct place but in corrupt state.
1038 # For local files write to the output directory not temporary dir.
1039 prefix = uri.dirname() if uri.isLocal else None
1040 with ResourcePath.temporary_uri(suffix=uri.getExtension(), prefix=prefix) as temporary_uri:
1041 # Need to configure the formatter to write to a different
1042 # location and that needs us to overwrite internals
1043 log.debug("Writing dataset to temporary location at %s", temporary_uri)
1045 # Assumes that if write_local_file is not subclassed that
1046 # to_bytes will be called using the base class definition.
1047 try:
1048 self.write_local_file(in_memory_dataset, temporary_uri)
1049 except Exception as e:
1050 e.add_note(
1051 f"Failed to serialize dataset {self.dataset_ref} of type"
1052 f" {get_full_type_name(in_memory_dataset)} to "
1053 f"temporary location {temporary_uri}."
1054 )
1055 raise
1057 # Use move for a local file since that becomes an efficient
1058 # os.rename. For remote resources we use copy to allow the
1059 # file to be cached afterwards.
1060 transfer = "move" if uri.isLocal else "copy"
1062 uri.transfer_from(temporary_uri, transfer=transfer, overwrite=True)
1064 if transfer == "copy":
1065 # Cache if required
1066 cache_manager.move_to_cache(temporary_uri, self._get_cache_ref())
1068 log.debug("Successfully wrote dataset to %s via a temporary file.", uri)
1070 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
1071 """Serialize the in-memory dataset to a local file.
1073 Parameters
1074 ----------
1075 in_memory_dataset : `object`
1076 The Python object to serialize.
1077 uri : `ResourcePath`
1078 The URI to use when writing the file.
1080 Notes
1081 -----
1082 By default this method will attempt to call `to_bytes` and then
1083 write these bytes to the file.
1085 Raises
1086 ------
1087 FormatterNotImplementedError
1088 Raised if the formatter subclass has not implemented this method
1089 or has failed to implement the `to_bytes` method.
1090 """
1091 log.debug("Writing bytes directly to %s", uri)
1092 uri.write(self.to_bytes(in_memory_dataset))
1093 log.debug("Successfully wrote bytes directly to %s", uri)
1095 def to_bytes(self, in_memory_dataset: Any) -> bytes:
1096 """Serialize the in-memory dataset to bytes.
1098 Parameters
1099 ----------
1100 in_memory_dataset : `object`
1101 The Python object to serialize.
1103 Returns
1104 -------
1105 serialized_dataset : `bytes`
1106 Bytes representing the serialized dataset.
1108 Raises
1109 ------
1110 FormatterNotImplementedError
1111 Raised if the formatter has not implemented the method. This will
1112 not cause a problem if `write_local_file` has been implemented.
1113 """
1114 raise FormatterNotImplementedError(
1115 f"This formatter can not convert {get_full_type_name(in_memory_dataset)} directly to bytes."
1116 )
1118 def make_updated_location(self, location: Location) -> Location:
1119 """Return a new `Location` updated with this formatter's extension.
1121 Parameters
1122 ----------
1123 location : `Location`
1124 The location to update.
1126 Returns
1127 -------
1128 updated : `Location`
1129 A new `Location` with a new file extension applied.
1130 """
1131 location = location.clone()
1132 # If the extension is "" the extension will be removed.
1133 location.updateExtension(self.get_write_extension())
1134 return location
1136 @classmethod
1137 def validate_extension(cls, location: Location) -> None:
1138 """Check the extension of the provided location for compatibility.
1140 Parameters
1141 ----------
1142 location : `Location`
1143 Location from which to extract a file extension.
1145 Raises
1146 ------
1147 ValueError
1148 Raised if the formatter does not understand this extension.
1149 """
1150 supported = set(cls.supported_extensions)
1151 default = cls.default_extension # type: ignore
1153 # If extension is implemented as an instance property it won't return
1154 # a string when called as a class property. Assume that
1155 # the supported extensions class property is complete.
1156 if default is not None and isinstance(default, str):
1157 supported.add(default)
1159 # Get the file name from the uri
1160 file = location.uri.basename()
1162 # Check that this file name ends with one of the supported extensions.
1163 # This is less prone to confusion than asking the location for
1164 # its extension and then doing a set comparison
1165 for ext in supported:
1166 if file.endswith(ext):
1167 return
1169 raise ValueError(
1170 f"Extension '{location.getExtension()}' on '{location}' "
1171 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})"
1172 )
1174 def predict_path(self) -> str:
1175 """Return the path that would be returned by write.
1177 Does not write any data file.
1179 Uses the `FileDescriptor` associated with the instance.
1181 Returns
1182 -------
1183 path : `str`
1184 Path within datastore that would be associated with the location
1185 stored in this `Formatter`.
1186 """
1187 updated = self.make_updated_location(self.file_descriptor.location)
1188 return updated.pathInStore.path
1190 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
1191 """Segregate the supplied parameters.
1193 This splits the parameters into those understood by the
1194 formatter and those not understood by the formatter.
1196 Any unsupported parameters are assumed to be usable by associated
1197 assemblers.
1199 Parameters
1200 ----------
1201 parameters : `dict`, optional
1202 Parameters with values that have been supplied by the caller
1203 and which might be relevant for the formatter. If `None`
1204 parameters will be read from the registered `FileDescriptor`.
1206 Returns
1207 -------
1208 supported : `dict`
1209 Those parameters supported by this formatter.
1210 unsupported : `dict`
1211 Those parameters not supported by this formatter.
1212 """
1213 if parameters is None:
1214 parameters = self.file_descriptor.parameters
1216 if parameters is None:
1217 return {}, {}
1219 if self.unsupported_parameters is None:
1220 # Support none of the parameters
1221 return {}, parameters.copy()
1223 # Start by assuming all are supported
1224 supported = parameters.copy()
1225 unsupported = {}
1227 # And remove any we know are not supported
1228 for p in set(supported):
1229 if p in self.unsupported_parameters:
1230 unsupported[p] = supported.pop(p)
1232 return supported, unsupported
1235class Formatter(metaclass=ABCMeta):
1236 """Interface for reading and writing Datasets.
1238 The formatters are associated with a particular `StorageClass`.
1240 Parameters
1241 ----------
1242 fileDescriptor : `FileDescriptor`, optional
1243 Identifies the file to read or write, and the associated storage
1244 classes and parameter information.
1245 dataId : `DataCoordinate`
1246 Data ID associated with this formatter.
1247 writeParameters : `dict`, optional
1248 Parameters to control how the dataset is serialized.
1249 writeRecipes : `dict`, optional
1250 Detailed write recipes indexed by recipe name.
1251 **kwargs
1252 Additional parameters that can allow parameters
1253 from `FormatterV2` to be provided.
1255 Notes
1256 -----
1257 All Formatter subclasses should share the base class's constructor
1258 signature.
1259 """
1261 # Now assuming that Formatter v1 can only refer to files so can add
1262 # this property for compatibility with v2.
1263 extension: str | None = None
1264 """Default file extension to use for writing files. None means that no
1265 modifications will be made to the supplied file extension. (`str`)"""
1267 unsupportedParameters: ClassVar[Set[str] | None] = frozenset()
1268 """Set of read parameters not understood by this `Formatter`. An empty set
1269 means all parameters are supported. `None` indicates that no parameters
1270 are supported. These parameters should match those defined in the storage
1271 class definition. (`frozenset`).
1272 """
1274 supportedWriteParameters: ClassVar[Set[str] | None] = None
1275 """Parameters understood by this formatter that can be used to control
1276 how a dataset is serialized. `None` indicates that no parameters are
1277 supported."""
1279 supportedExtensions: ClassVar[Set[str]] = frozenset()
1280 """Set of all extensions supported by this formatter.
1282 Only expected to be populated by Formatters that write files. Any extension
1283 assigned to the ``extension`` property will be automatically included in
1284 the list of supported extensions."""
1286 def __init__(
1287 self,
1288 fileDescriptor: FileDescriptor,
1289 *,
1290 dataId: DataCoordinate | None = None,
1291 writeParameters: Mapping[str, Any] | None = None,
1292 writeRecipes: Mapping[str, Any] | None = None,
1293 # Allow FormatterV2 parameters to be dropped.
1294 **kwargs: Any,
1295 ):
1296 if not isinstance(fileDescriptor, FileDescriptor):
1297 raise TypeError("File descriptor must be a FileDescriptor")
1298 self._fileDescriptor = fileDescriptor
1300 if dataId is None:
1301 raise RuntimeError("dataId is now required for formatter initialization")
1302 if not isinstance(dataId, DataCoordinate):
1303 raise TypeError(f"DataId is required to be a DataCoordinate but got {type(dataId)}.")
1304 self._dataId = dataId
1306 # V2 compatibility.
1307 if writeParameters is None:
1308 writeParameters = kwargs.get("write_parameters")
1309 if writeRecipes is None:
1310 writeRecipes = kwargs.get("write_recipes")
1312 # Check that the write parameters are allowed
1313 if writeParameters:
1314 if self.supportedWriteParameters is None:
1315 raise ValueError(
1316 f"This formatter does not accept any write parameters. Got: {', '.join(writeParameters)}"
1317 )
1318 else:
1319 given = set(writeParameters)
1320 unknown = given - self.supportedWriteParameters
1321 if unknown:
1322 s = "s" if len(unknown) != 1 else ""
1323 unknownStr = ", ".join(f"'{u}'" for u in unknown)
1324 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}")
1326 self._writeParameters = writeParameters
1327 self._writeRecipes = self.validate_write_recipes(writeRecipes)
1329 def __str__(self) -> str:
1330 return f"{self.name()}@{self.fileDescriptor.location.path}"
1332 def __repr__(self) -> str:
1333 return f"{self.name()}({self.fileDescriptor!r})"
1335 @property
1336 def fileDescriptor(self) -> FileDescriptor:
1337 """File descriptor associated with this formatter
1338 (`FileDescriptor`).
1339 """
1340 return self._fileDescriptor
1342 @property
1343 def dataId(self) -> DataCoordinate:
1344 """Return Data ID associated with this formatter (`DataCoordinate`)."""
1345 return self._dataId
1347 @property
1348 def writeParameters(self) -> Mapping[str, Any]:
1349 """Parameters to use when writing out datasets."""
1350 if self._writeParameters is not None:
1351 return self._writeParameters
1352 return {}
1354 @property
1355 def writeRecipes(self) -> Mapping[str, Any]:
1356 """Detailed write Recipes indexed by recipe name."""
1357 if self._writeRecipes is not None:
1358 return self._writeRecipes
1359 return {}
1361 def can_accept(self, in_memory_dataset: Any) -> bool:
1362 """Indicate whether this formatter can accept the specified
1363 storage class directly.
1365 Parameters
1366 ----------
1367 in_memory_dataset : `object`
1368 The dataset that is to be written.
1370 Returns
1371 -------
1372 accepts : `bool`
1373 If `True` the formatter can write data of this type without
1374 requiring datastore to convert it. If `False` the datastore
1375 will attempt to convert before writing.
1377 Notes
1378 -----
1379 The base class checks that the given python type matches
1380 the python type specified for this formatter when
1381 constructed.
1382 """
1383 return isinstance(in_memory_dataset, self.file_descriptor.storageClass.pytype)
1385 @classmethod
1386 def validateWriteRecipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
1387 """Validate supplied recipes for this formatter.
1389 The recipes are supplemented with default values where appropriate.
1391 Parameters
1392 ----------
1393 recipes : `dict`
1394 Recipes to validate.
1396 Returns
1397 -------
1398 validated : `dict`
1399 Validated recipes.
1401 Raises
1402 ------
1403 RuntimeError
1404 Raised if validation fails. The default implementation raises
1405 if any recipes are given.
1406 """
1407 if recipes:
1408 raise RuntimeError(f"This formatter does not understand these writeRecipes: {recipes}")
1409 return recipes
1411 @classmethod
1412 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
1413 return cls.validateWriteRecipes(recipes)
1415 @classmethod
1416 def name(cls) -> str:
1417 """Return the fully qualified name of the formatter.
1419 Returns
1420 -------
1421 name : `str`
1422 Fully-qualified name of formatter class.
1423 """
1424 return get_full_type_name(cls)
1426 @abstractmethod
1427 def read(self, component: str | None = None) -> Any:
1428 """Read a Dataset.
1430 Parameters
1431 ----------
1432 component : `str`, optional
1433 Component to read from the file. Only used if the `StorageClass`
1434 for reading differed from the `StorageClass` used to write the
1435 file.
1437 Returns
1438 -------
1439 inMemoryDataset : `object`
1440 The requested Dataset.
1441 """
1442 raise FormatterNotImplementedError("Type does not support reading")
1444 @abstractmethod
1445 def write(self, inMemoryDataset: Any) -> None:
1446 """Write a Dataset.
1448 Parameters
1449 ----------
1450 inMemoryDataset : `object`
1451 The Dataset to store.
1452 """
1453 raise FormatterNotImplementedError("Type does not support writing")
1455 @classmethod
1456 def can_read_bytes(cls) -> bool:
1457 """Indicate if this formatter can format from bytes.
1459 Returns
1460 -------
1461 can : `bool`
1462 `True` if the `fromBytes` method is implemented.
1463 """
1464 # We have no property to read so instead try to format from a byte
1465 # and see what happens
1466 try:
1467 # We know the arguments are incompatible
1468 cls.fromBytes(cls, b"") # type: ignore
1469 except FormatterNotImplementedError:
1470 return False
1471 except Exception:
1472 # There will be problems with the bytes we are supplying so ignore
1473 pass
1474 return True
1476 def fromBytes(self, serializedDataset: bytes, component: str | None = None) -> object:
1477 """Read serialized data into a Dataset or its component.
1479 Parameters
1480 ----------
1481 serializedDataset : `bytes`
1482 Bytes object to unserialize.
1483 component : `str`, optional
1484 Component to read from the Dataset. Only used if the `StorageClass`
1485 for reading differed from the `StorageClass` used to write the
1486 file.
1488 Returns
1489 -------
1490 inMemoryDataset : `object`
1491 The requested data as a Python object. The type of object
1492 is controlled by the specific formatter.
1493 """
1494 raise FormatterNotImplementedError("Type does not support reading from bytes.")
1496 def toBytes(self, inMemoryDataset: Any) -> bytes:
1497 """Serialize the Dataset to bytes based on formatter.
1499 Parameters
1500 ----------
1501 inMemoryDataset : `object`
1502 The Python object to serialize.
1504 Returns
1505 -------
1506 serializedDataset : `bytes`
1507 Bytes representing the serialized dataset.
1508 """
1509 raise FormatterNotImplementedError("Type does not support writing to bytes.")
1511 @contextlib.contextmanager
1512 def _updateLocation(self, location: Location | None) -> Iterator[Location]:
1513 """Temporarily replace the location associated with this formatter.
1515 Parameters
1516 ----------
1517 location : `Location`
1518 New location to use for this formatter. If `None` the
1519 formatter will not change but it will still return
1520 the old location. This allows it to be used in a code
1521 path where the location may not need to be updated
1522 but the with block is still convenient.
1524 Yields
1525 ------
1526 old : `Location`
1527 The old location that will be restored.
1529 Notes
1530 -----
1531 This is an internal method that should be used with care.
1532 It may change in the future. Should be used as a context
1533 manager to restore the location when the temporary is no
1534 longer required.
1535 """
1536 old = self._fileDescriptor.location
1537 try:
1538 if location is not None:
1539 self._fileDescriptor.location = location
1540 yield old
1541 finally:
1542 if location is not None:
1543 self._fileDescriptor.location = old
1545 def make_updated_location(self, location: Location) -> Location:
1546 """Return a new `Location` updated with this formatter's extension.
1548 Parameters
1549 ----------
1550 location : `Location`
1551 The location to update.
1553 Returns
1554 -------
1555 updated : `Location`
1556 A new `Location` with a new file extension applied.
1558 Raises
1559 ------
1560 NotImplementedError
1561 Raised if there is no ``extension`` attribute associated with
1562 this formatter.
1564 Notes
1565 -----
1566 This method is available to all Formatters but might not be
1567 implemented by all formatters. It requires that a formatter set
1568 an ``extension`` attribute containing the file extension used when
1569 writing files. If ``extension`` is `None` the supplied file will
1570 not be updated. Not all formatters write files so this is not
1571 defined in the base class.
1572 """
1573 location = location.clone()
1574 try:
1575 # We are deliberately allowing extension to be undefined by
1576 # default in the base class and mypy complains.
1577 location.updateExtension(self.extension) # type:ignore
1578 except AttributeError:
1579 raise NotImplementedError("No file extension registered with this formatter") from None
1580 return location
1582 @classmethod
1583 def validate_extension(cls, location: Location) -> None:
1584 """Check the extension of the provided location for compatibility.
1586 Parameters
1587 ----------
1588 location : `Location`
1589 Location from which to extract a file extension.
1591 Raises
1592 ------
1593 NotImplementedError
1594 Raised if file extensions are a concept not understood by this
1595 formatter.
1596 ValueError
1597 Raised if the formatter does not understand this extension.
1599 Notes
1600 -----
1601 This method is available to all Formatters but might not be
1602 implemented by all formatters. It requires that a formatter set
1603 an ``extension`` attribute containing the file extension used when
1604 writing files. If ``extension`` is `None` only the set of supported
1605 extensions will be examined.
1606 """
1607 supported = set(cls.supportedExtensions)
1609 try:
1610 # We are deliberately allowing extension to be undefined by
1611 # default in the base class and mypy complains.
1612 default = cls.extension # type: ignore
1613 except AttributeError:
1614 raise NotImplementedError("No file extension registered with this formatter") from None
1616 # If extension is implemented as an instance property it won't return
1617 # a string when called as a class property. Assume that
1618 # the supported extensions class property is complete.
1619 if default is not None and isinstance(default, str):
1620 supported.add(default)
1622 # Get the file name from the uri
1623 file = location.uri.basename()
1625 # Check that this file name ends with one of the supported extensions.
1626 # This is less prone to confusion than asking the location for
1627 # its extension and then doing a set comparison
1628 for ext in supported:
1629 if file.endswith(ext):
1630 return
1632 raise ValueError(
1633 f"Extension '{location.getExtension()}' on '{location}' "
1634 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})"
1635 )
1637 def predict_path(self) -> str:
1638 """Return the path that would be returned by write.
1640 Does not write any data file.
1642 Uses the `FileDescriptor` associated with the instance.
1644 Returns
1645 -------
1646 path : `str`
1647 Path within datastore that would be associated with the location
1648 stored in this `Formatter`.
1649 """
1650 updated = self.make_updated_location(self.fileDescriptor.location)
1651 return updated.pathInStore.path
1653 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
1654 """Segregate the supplied parameters.
1656 This splits the parameters into those understood by the
1657 formatter and those not understood by the formatter.
1659 Any unsupported parameters are assumed to be usable by associated
1660 assemblers.
1662 Parameters
1663 ----------
1664 parameters : `dict`, optional
1665 Parameters with values that have been supplied by the caller
1666 and which might be relevant for the formatter. If `None`
1667 parameters will be read from the registered `FileDescriptor`.
1669 Returns
1670 -------
1671 supported : `dict`
1672 Those parameters supported by this formatter.
1673 unsupported : `dict`
1674 Those parameters not supported by this formatter.
1675 """
1676 if parameters is None:
1677 parameters = self.fileDescriptor.parameters
1679 if parameters is None:
1680 return {}, {}
1682 if self.unsupportedParameters is None:
1683 # Support none of the parameters
1684 return {}, parameters.copy()
1686 # Start by assuming all are supported
1687 supported = parameters.copy()
1688 unsupported = {}
1690 # And remove any we know are not supported
1691 for p in set(supported):
1692 if p in self.unsupportedParameters:
1693 unsupported[p] = supported.pop(p)
1695 return supported, unsupported
1697 # Support classic V1 interface.
1698 makeUpdatedLocation = make_updated_location
1699 validateExtension = validate_extension
1700 segregateParameters = segregate_parameters
1701 predictPath = predict_path
1703 # Compatibility with V2 properties.
1704 @property
1705 def write_parameters(self) -> Mapping[str, Any]:
1706 return self.writeParameters
1708 @property
1709 def write_recipes(self) -> Mapping[str, Any]:
1710 return self.writeRecipes
1712 @property
1713 def file_descriptor(self) -> FileDescriptor:
1714 return self.fileDescriptor
1716 @property
1717 def data_id(self) -> DataCoordinate:
1718 return self.dataId
1721class FormatterFactory:
1722 """Factory for `Formatter` instances."""
1724 defaultKey = LookupKey("default")
1725 """Configuration key associated with default write parameter settings."""
1727 writeRecipesKey = LookupKey("write_recipes")
1728 """Configuration key associated with write recipes."""
1730 def __init__(self) -> None:
1731 self._mappingFactory = MappingFactory(Formatter)
1733 def __contains__(self, key: LookupKey | str) -> bool:
1734 """Indicate whether the supplied key is present in the factory.
1736 Parameters
1737 ----------
1738 key : `LookupKey`, `str` or objects with ``name`` attribute
1739 Key to use to lookup in the factory whether a corresponding
1740 formatter is present.
1742 Returns
1743 -------
1744 in : `bool`
1745 `True` if the supplied key is present in the factory.
1746 """
1747 return key in self._mappingFactory
1749 def registerFormatters(self, config: Config, *, universe: DimensionUniverse) -> None:
1750 """Bulk register formatters from a config.
1752 Parameters
1753 ----------
1754 config : `Config`
1755 ``formatters`` section of a configuration.
1756 universe : `DimensionUniverse`, optional
1757 Set of all known dimensions, used to expand and validate any used
1758 in lookup keys.
1760 Notes
1761 -----
1762 The configuration can include one level of hierarchy where an
1763 instrument-specific section can be defined to override more general
1764 template specifications. This is represented in YAML using a
1765 key of form ``instrument<name>`` which can then define templates
1766 that will be returned if a `DatasetRef` contains a matching instrument
1767 name in the data ID.
1769 The config is parsed using the function
1770 `~lsst.daf.butler.configSubset.processLookupConfigs`.
1772 The values for formatter entries can be either a simple string
1773 referring to a python type or a dict representing the formatter and
1774 parameters to be hard-coded into the formatter constructor. For
1775 the dict case the following keys are supported:
1777 - formatter: The python type to be used as the formatter class.
1778 - parameters: A further dict to be passed directly to the
1779 ``write_parameters`` Formatter constructor to seed it.
1780 These parameters are validated at instance creation and not at
1781 configuration.
1783 Additionally, a special ``default`` section can be defined that
1784 uses the formatter type (class) name as the keys and specifies
1785 default write parameters that should be used whenever an instance
1786 of that class is constructed.
1788 .. code-block:: yaml
1790 formatters:
1791 default:
1792 lsst.daf.butler.formatters.example.ExampleFormatter:
1793 max: 10
1794 min: 2
1795 comment: Default comment
1796 calexp: lsst.daf.butler.formatters.example.ExampleFormatter
1797 coadd:
1798 formatter: lsst.daf.butler.formatters.example.ExampleFormatter
1799 parameters:
1800 max: 5
1802 Any time an ``ExampleFormatter`` is constructed it will use those
1803 parameters. If an explicit entry later in the configuration specifies
1804 a different set of parameters, the two will be merged with the later
1805 entry taking priority. In the example above ``calexp`` will use
1806 the default parameters but ``coadd`` will override the value for
1807 ``max``.
1809 Formatter configuration can also include a special section describing
1810 collections of write parameters that can be accessed through a
1811 simple label. This allows common collections of options to be
1812 specified in one place in the configuration and reused later.
1813 The ``write_recipes`` section is indexed by Formatter class name
1814 and each key is the label to associate with the parameters.
1816 .. code-block:: yaml
1818 formatters:
1819 write_recipes:
1820 lsst.obs.base.formatters.fitsExposure.FixExposureFormatter:
1821 lossless:
1822 ...
1823 noCompression:
1824 ...
1826 By convention a formatter that uses write recipes will support a
1827 ``recipe`` write parameter that will refer to a recipe name in
1828 the ``write_recipes`` component. The `Formatter` will be constructed
1829 in the `FormatterFactory` with all the relevant recipes and
1830 will not attempt to filter by looking at ``write_parameters`` in
1831 advance. See the specific formatter documentation for details on
1832 acceptable recipe options.
1833 """
1834 allowed_keys = {"formatter", "parameters"}
1836 contents = processLookupConfigs(config, allow_hierarchy=True, universe=universe)
1838 # Extract any default parameter settings
1839 defaultParameters = contents.get(self.defaultKey, {})
1840 if not isinstance(defaultParameters, Mapping):
1841 raise RuntimeError(
1842 "Default formatter parameters in config can not be a single string"
1843 f" (got: {type(defaultParameters)})"
1844 )
1846 # Extract any global write recipes -- these are indexed by
1847 # Formatter class name.
1848 writeRecipes = contents.get(self.writeRecipesKey, {})
1849 if isinstance(writeRecipes, str):
1850 raise RuntimeError(
1851 f"The formatters.{self.writeRecipesKey} section must refer to a dict not '{writeRecipes}'"
1852 )
1854 for key, f in contents.items():
1855 # default is handled in a special way
1856 if key == self.defaultKey:
1857 continue
1858 if key == self.writeRecipesKey:
1859 continue
1861 # Can be a str or a dict.
1862 specificWriteParameters = {}
1863 if isinstance(f, str):
1864 formatter = f
1865 elif isinstance(f, Mapping):
1866 all_keys = set(f)
1867 unexpected_keys = all_keys - allowed_keys
1868 if unexpected_keys:
1869 raise ValueError(f"Formatter {key} uses unexpected keys {unexpected_keys} in config")
1870 if "formatter" not in f:
1871 raise ValueError(f"Mandatory 'formatter' key missing for formatter key {key}")
1872 formatter = f["formatter"]
1873 if "parameters" in f:
1874 specificWriteParameters = f["parameters"]
1875 else:
1876 raise ValueError(f"Formatter for key {key} has unexpected value: '{f}'")
1878 # Apply any default parameters for this formatter
1879 writeParameters = copy.deepcopy(defaultParameters.get(formatter, {}))
1880 writeParameters.update(specificWriteParameters)
1882 kwargs: dict[str, Any] = {}
1883 if writeParameters:
1884 kwargs["write_parameters"] = writeParameters
1886 if formatter in writeRecipes:
1887 kwargs["write_recipes"] = writeRecipes[formatter]
1889 self.registerFormatter(key, formatter, **kwargs)
1891 def getLookupKeys(self) -> set[LookupKey]:
1892 """Retrieve the look up keys for all the registry entries.
1894 Returns
1895 -------
1896 keys : `set` of `LookupKey`
1897 The keys available for matching in the registry.
1898 """
1899 return self._mappingFactory.getLookupKeys()
1901 def getFormatterClassWithMatch(
1902 self, entity: Entity
1903 ) -> tuple[LookupKey, type[Formatter | FormatterV2], dict[str, Any]]:
1904 """Get the matching formatter class along with the registry key.
1906 Parameters
1907 ----------
1908 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1909 Entity to use to determine the formatter to return.
1910 `StorageClass` will be used as a last resort if `DatasetRef`
1911 or `DatasetType` instance is provided. Supports instrument
1912 override if a `DatasetRef` is provided configured with an
1913 ``instrument`` value for the data ID.
1915 Returns
1916 -------
1917 matchKey : `LookupKey`
1918 The key that resulted in the successful match.
1919 formatter : `type`
1920 The class of the registered formatter.
1921 formatter_kwargs : `dict`
1922 Keyword arguments that are associated with this formatter entry.
1923 """
1924 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
1925 matchKey, formatter, formatter_kwargs = self._mappingFactory.getClassFromRegistryWithMatch(names)
1926 log.debug(
1927 "Retrieved formatter %s from key '%s' for entity '%s'",
1928 get_full_type_name(formatter),
1929 matchKey,
1930 entity,
1931 )
1933 return matchKey, formatter, formatter_kwargs
1935 def getFormatterClass(self, entity: Entity) -> type:
1936 """Get the matching formatter class.
1938 Parameters
1939 ----------
1940 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1941 Entity to use to determine the formatter to return.
1942 `StorageClass` will be used as a last resort if `DatasetRef`
1943 or `DatasetType` instance is provided. Supports instrument
1944 override if a `DatasetRef` is provided configured with an
1945 ``instrument`` value for the data ID.
1947 Returns
1948 -------
1949 formatter : `type`
1950 The class of the registered formatter.
1951 """
1952 _, formatter, _ = self.getFormatterClassWithMatch(entity)
1953 return formatter
1955 def getFormatterWithMatch(
1956 self, entity: Entity, *args: Any, **kwargs: Any
1957 ) -> tuple[LookupKey, Formatter | FormatterV2]:
1958 """Get a new formatter instance along with the matching registry key.
1960 Parameters
1961 ----------
1962 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1963 Entity to use to determine the formatter to return.
1964 `StorageClass` will be used as a last resort if `DatasetRef`
1965 or `DatasetType` instance is provided. Supports instrument
1966 override if a `DatasetRef` is provided configured with an
1967 ``instrument`` value for the data ID.
1968 *args : `tuple`
1969 Positional arguments to use pass to the object constructor.
1970 **kwargs
1971 Keyword arguments to pass to object constructor.
1973 Returns
1974 -------
1975 matchKey : `LookupKey`
1976 The key that resulted in the successful match.
1977 formatter : `Formatter`
1978 An instance of the registered formatter.
1979 """
1980 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
1981 matchKey, formatter = self._mappingFactory.getFromRegistryWithMatch(names, *args, **kwargs)
1982 log.debug(
1983 "Retrieved formatter %s from key '%s' for entity '%s'",
1984 get_full_type_name(formatter),
1985 matchKey,
1986 entity,
1987 )
1989 return matchKey, formatter
1991 def getFormatter(self, entity: Entity, *args: Any, **kwargs: Any) -> Formatter | FormatterV2:
1992 """Get a new formatter instance.
1994 Parameters
1995 ----------
1996 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1997 Entity to use to determine the formatter to return.
1998 `StorageClass` will be used as a last resort if `DatasetRef`
1999 or `DatasetType` instance is provided. Supports instrument
2000 override if a `DatasetRef` is provided configured with an
2001 ``instrument`` value for the data ID.
2002 *args : `tuple`
2003 Positional arguments to use pass to the object constructor.
2004 **kwargs
2005 Keyword arguments to pass to object constructor.
2007 Returns
2008 -------
2009 formatter : `Formatter`
2010 An instance of the registered formatter.
2011 """
2012 _, formatter = self.getFormatterWithMatch(entity, *args, **kwargs)
2013 return formatter
2015 def registerFormatter(
2016 self,
2017 type_: LookupKey | str | StorageClass | DatasetType,
2018 formatter: str,
2019 *,
2020 overwrite: bool = False,
2021 **kwargs: Any,
2022 ) -> None:
2023 """Register a `Formatter`.
2025 Parameters
2026 ----------
2027 type_ : `LookupKey`, `str`, `StorageClass` or `DatasetType`
2028 Type for which this formatter is to be used. If a `LookupKey`
2029 is not provided, one will be constructed from the supplied string
2030 or by using the ``name`` property of the supplied entity.
2031 formatter : `str` or class of type `Formatter`
2032 Identifies a `Formatter` subclass to use for reading and writing
2033 Datasets of this type. Can be a `Formatter` class.
2034 overwrite : `bool`, optional
2035 If `True` an existing entry will be replaced by the new value.
2036 Default is `False`.
2037 **kwargs
2038 Keyword arguments to always pass to object constructor when
2039 retrieved.
2041 Raises
2042 ------
2043 ValueError
2044 Raised if the formatter does not name a valid formatter type and
2045 ``overwrite`` is `False`.
2046 """
2047 self._mappingFactory.placeInRegistry(type_, formatter, overwrite=overwrite, **kwargs)
2050class FormatterV1inV2(FormatterV2):
2051 """An implementation of a V2 formatter that provides a compatibility
2052 interface for V1 formatters.
2054 Parameters
2055 ----------
2056 file_descriptor : `FileDescriptor`, optional
2057 Identifies the file to read or write, and the associated storage
2058 classes and parameter information. Its value can be `None` if the
2059 caller will never call `Formatter.read` or `Formatter.write`.
2060 ref : `DatasetRef`
2061 The dataset associated with this formatter. Should not be a component
2062 dataset ref.
2063 formatter : `Formatter`
2064 A version 1 `Formatter` instance. The V2 formatter layer forwards calls
2065 to this formatter.
2066 write_parameters : `dict`, optional
2067 Any parameters to be hard-coded into this instance to control how
2068 the dataset is serialized.
2069 write_recipes : `dict`, optional
2070 Detailed write Recipes indexed by recipe name.
2071 **kwargs
2072 Additional arguments that will be ignored but allow for
2073 `Formatter` V1 parameters to be given.
2074 """
2076 can_read_from_local_file = True
2077 """This formatter can read from a local file."""
2079 def __init__(
2080 self,
2081 file_descriptor: FileDescriptor,
2082 *,
2083 ref: DatasetRef,
2084 formatter: Formatter,
2085 write_parameters: Mapping[str, Any] | None = None,
2086 write_recipes: Mapping[str, Any] | None = None,
2087 # Compatibility parameters. Unused in v2.
2088 **kwargs: Any,
2089 ):
2090 if not isinstance(formatter, Formatter):
2091 raise TypeError(f"Formatter parameter was not a V1 formatter (was {type(formatter)})")
2093 # Replace the class property with instance values from this
2094 # V1 formatter so that the V2 __init__ will be able to validate.
2095 self.supported_write_parameters = formatter.supportedWriteParameters # type: ignore
2096 self._formatter = formatter
2098 super().__init__(
2099 file_descriptor, ref=ref, write_parameters=write_parameters, write_recipes=write_recipes
2100 )
2102 def get_write_extension(self) -> str:
2103 ext = self._formatter.extension
2104 return ext if ext is not None else ""
2106 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
2107 return self._formatter.segregate_parameters(parameters)
2109 def validate_write_recipes( # type: ignore
2110 self,
2111 recipes: Mapping[str, Any] | None,
2112 ) -> Mapping[str, Any] | None:
2113 # This should be a class method but a class method can not work
2114 # for a dynamic shim. Luckily the shim is only used as an instance.
2115 return self._formatter.validate_write_recipes(recipes)
2117 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any:
2118 # Need to temporarily override the location since the V1 formatter
2119 # will not know anything about this local file.
2121 # V2 does not have a fromBytes equivalent.
2122 if self._formatter.can_read_bytes():
2123 with open(path, "rb") as fd:
2124 serialized_dataset = fd.read()
2125 return self._formatter.fromBytes(serialized_dataset, component=component)
2127 location = Location(None, path)
2128 with self._formatter._updateLocation(location):
2129 try:
2130 result = self._formatter.read(component=component)
2131 except NotImplementedError:
2132 # V1 raises NotImplementedError but V2 is expecting something
2133 # slightly different.
2134 return NotImplemented
2135 return result
2137 def to_bytes(self, in_memory_dataset: Any) -> bytes:
2138 try:
2139 return self._formatter.toBytes(in_memory_dataset)
2140 except NotImplementedError as e:
2141 # V1 raises NotImplementedError but V2 is expecting something
2142 # slightly different.
2143 raise FormatterNotImplementedError(str(e)) from e
2145 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
2146 with self._formatter._updateLocation(Location(None, uri)):
2147 try:
2148 self._formatter.write(in_memory_dataset)
2149 except NotImplementedError as e:
2150 # V1 raises NotImplementedError but V2 is expecting something
2151 # slightly different.
2152 raise FormatterNotImplementedError(str(e)) from e
2155# Type to use when allowing a Formatter or its class name
2156FormatterParameter: TypeAlias = str | type[Formatter] | Formatter | FormatterV2 | type[FormatterV2]