Coverage for python / lsst / daf / butler / _formatter.py: 23%
533 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-17 16:22 -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 if prefix is not None:
1041 prefix.mkdir()
1042 with ResourcePath.temporary_uri(suffix=uri.getExtension(), prefix=prefix) as temporary_uri:
1043 # Need to configure the formatter to write to a different
1044 # location and that needs us to overwrite internals
1045 log.debug("Writing dataset to temporary location at %s", temporary_uri)
1047 # Assumes that if write_local_file is not subclassed that
1048 # to_bytes will be called using the base class definition.
1049 try:
1050 self.write_local_file(in_memory_dataset, temporary_uri)
1051 except Exception as e:
1052 e.add_note(
1053 f"Failed to serialize dataset {self.dataset_ref} of type"
1054 f" {get_full_type_name(in_memory_dataset)} to "
1055 f"temporary location {temporary_uri}."
1056 )
1057 raise
1059 # Use move for a local file since that becomes an efficient
1060 # os.rename. For remote resources we use copy to allow the
1061 # file to be cached afterwards.
1062 transfer = "move" if uri.isLocal else "copy"
1064 uri.transfer_from(temporary_uri, transfer=transfer, overwrite=True)
1066 if transfer == "copy":
1067 # Cache if required
1068 cache_manager.move_to_cache(temporary_uri, self._get_cache_ref())
1070 log.debug("Successfully wrote dataset to %s via a temporary file.", uri)
1072 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
1073 """Serialize the in-memory dataset to a local file.
1075 Parameters
1076 ----------
1077 in_memory_dataset : `object`
1078 The Python object to serialize.
1079 uri : `ResourcePath`
1080 The URI to use when writing the file.
1082 Notes
1083 -----
1084 By default this method will attempt to call `to_bytes` and then
1085 write these bytes to the file.
1087 Raises
1088 ------
1089 FormatterNotImplementedError
1090 Raised if the formatter subclass has not implemented this method
1091 or has failed to implement the `to_bytes` method.
1092 """
1093 log.debug("Writing bytes directly to %s", uri)
1094 uri.write(self.to_bytes(in_memory_dataset))
1095 log.debug("Successfully wrote bytes directly to %s", uri)
1097 def to_bytes(self, in_memory_dataset: Any) -> bytes:
1098 """Serialize the in-memory dataset to bytes.
1100 Parameters
1101 ----------
1102 in_memory_dataset : `object`
1103 The Python object to serialize.
1105 Returns
1106 -------
1107 serialized_dataset : `bytes`
1108 Bytes representing the serialized dataset.
1110 Raises
1111 ------
1112 FormatterNotImplementedError
1113 Raised if the formatter has not implemented the method. This will
1114 not cause a problem if `write_local_file` has been implemented.
1115 """
1116 raise FormatterNotImplementedError(
1117 f"This formatter can not convert {get_full_type_name(in_memory_dataset)} directly to bytes."
1118 )
1120 def make_updated_location(self, location: Location) -> Location:
1121 """Return a new `Location` updated with this formatter's extension.
1123 Parameters
1124 ----------
1125 location : `Location`
1126 The location to update.
1128 Returns
1129 -------
1130 updated : `Location`
1131 A new `Location` with a new file extension applied.
1132 """
1133 location = location.clone()
1134 # If the extension is "" the extension will be removed.
1135 location.updateExtension(self.get_write_extension())
1136 return location
1138 @classmethod
1139 def validate_extension(cls, location: Location) -> None:
1140 """Check the extension of the provided location for compatibility.
1142 Parameters
1143 ----------
1144 location : `Location`
1145 Location from which to extract a file extension.
1147 Raises
1148 ------
1149 ValueError
1150 Raised if the formatter does not understand this extension.
1151 """
1152 supported = set(cls.supported_extensions)
1153 default = cls.default_extension # type: ignore
1155 # If extension is implemented as an instance property it won't return
1156 # a string when called as a class property. Assume that
1157 # the supported extensions class property is complete.
1158 if default is not None and isinstance(default, str):
1159 supported.add(default)
1161 # Get the file name from the uri
1162 file = location.uri.basename()
1164 # Check that this file name ends with one of the supported extensions.
1165 # This is less prone to confusion than asking the location for
1166 # its extension and then doing a set comparison
1167 for ext in supported:
1168 if file.endswith(ext):
1169 return
1171 raise ValueError(
1172 f"Extension '{location.getExtension()}' on '{location}' "
1173 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})"
1174 )
1176 def predict_path(self) -> str:
1177 """Return the path that would be returned by write.
1179 Does not write any data file.
1181 Uses the `FileDescriptor` associated with the instance.
1183 Returns
1184 -------
1185 path : `str`
1186 Path within datastore that would be associated with the location
1187 stored in this `Formatter`.
1188 """
1189 updated = self.make_updated_location(self.file_descriptor.location)
1190 return updated.pathInStore.path
1192 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
1193 """Segregate the supplied parameters.
1195 This splits the parameters into those understood by the
1196 formatter and those not understood by the formatter.
1198 Any unsupported parameters are assumed to be usable by associated
1199 assemblers.
1201 Parameters
1202 ----------
1203 parameters : `dict`, optional
1204 Parameters with values that have been supplied by the caller
1205 and which might be relevant for the formatter. If `None`
1206 parameters will be read from the registered `FileDescriptor`.
1208 Returns
1209 -------
1210 supported : `dict`
1211 Those parameters supported by this formatter.
1212 unsupported : `dict`
1213 Those parameters not supported by this formatter.
1214 """
1215 if parameters is None:
1216 parameters = self.file_descriptor.parameters
1218 if parameters is None:
1219 return {}, {}
1221 if self.unsupported_parameters is None:
1222 # Support none of the parameters
1223 return {}, parameters.copy()
1225 # Start by assuming all are supported
1226 supported = parameters.copy()
1227 unsupported = {}
1229 # And remove any we know are not supported
1230 for p in set(supported):
1231 if p in self.unsupported_parameters:
1232 unsupported[p] = supported.pop(p)
1234 return supported, unsupported
1237class Formatter(metaclass=ABCMeta):
1238 """Interface for reading and writing Datasets.
1240 The formatters are associated with a particular `StorageClass`.
1242 Parameters
1243 ----------
1244 fileDescriptor : `FileDescriptor`, optional
1245 Identifies the file to read or write, and the associated storage
1246 classes and parameter information.
1247 dataId : `DataCoordinate`
1248 Data ID associated with this formatter.
1249 writeParameters : `dict`, optional
1250 Parameters to control how the dataset is serialized.
1251 writeRecipes : `dict`, optional
1252 Detailed write recipes indexed by recipe name.
1253 **kwargs
1254 Additional parameters that can allow parameters
1255 from `FormatterV2` to be provided.
1257 Notes
1258 -----
1259 All Formatter subclasses should share the base class's constructor
1260 signature.
1261 """
1263 # Now assuming that Formatter v1 can only refer to files so can add
1264 # this property for compatibility with v2.
1265 extension: str | None = None
1266 """Default file extension to use for writing files. None means that no
1267 modifications will be made to the supplied file extension. (`str`)"""
1269 unsupportedParameters: ClassVar[Set[str] | None] = frozenset()
1270 """Set of read parameters not understood by this `Formatter`. An empty set
1271 means all parameters are supported. `None` indicates that no parameters
1272 are supported. These parameters should match those defined in the storage
1273 class definition. (`frozenset`).
1274 """
1276 supportedWriteParameters: ClassVar[Set[str] | None] = None
1277 """Parameters understood by this formatter that can be used to control
1278 how a dataset is serialized. `None` indicates that no parameters are
1279 supported."""
1281 supportedExtensions: ClassVar[Set[str]] = frozenset()
1282 """Set of all extensions supported by this formatter.
1284 Only expected to be populated by Formatters that write files. Any extension
1285 assigned to the ``extension`` property will be automatically included in
1286 the list of supported extensions."""
1288 def __init__(
1289 self,
1290 fileDescriptor: FileDescriptor,
1291 *,
1292 dataId: DataCoordinate | None = None,
1293 writeParameters: Mapping[str, Any] | None = None,
1294 writeRecipes: Mapping[str, Any] | None = None,
1295 # Allow FormatterV2 parameters to be dropped.
1296 **kwargs: Any,
1297 ):
1298 if not isinstance(fileDescriptor, FileDescriptor):
1299 raise TypeError("File descriptor must be a FileDescriptor")
1300 self._fileDescriptor = fileDescriptor
1302 if dataId is None:
1303 raise RuntimeError("dataId is now required for formatter initialization")
1304 if not isinstance(dataId, DataCoordinate):
1305 raise TypeError(f"DataId is required to be a DataCoordinate but got {type(dataId)}.")
1306 self._dataId = dataId
1308 # V2 compatibility.
1309 if writeParameters is None:
1310 writeParameters = kwargs.get("write_parameters")
1311 if writeRecipes is None:
1312 writeRecipes = kwargs.get("write_recipes")
1314 # Check that the write parameters are allowed
1315 if writeParameters:
1316 if self.supportedWriteParameters is None:
1317 raise ValueError(
1318 f"This formatter does not accept any write parameters. Got: {', '.join(writeParameters)}"
1319 )
1320 else:
1321 given = set(writeParameters)
1322 unknown = given - self.supportedWriteParameters
1323 if unknown:
1324 s = "s" if len(unknown) != 1 else ""
1325 unknownStr = ", ".join(f"'{u}'" for u in unknown)
1326 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}")
1328 self._writeParameters = writeParameters
1329 self._writeRecipes = self.validate_write_recipes(writeRecipes)
1331 def __str__(self) -> str:
1332 return f"{self.name()}@{self.fileDescriptor.location.path}"
1334 def __repr__(self) -> str:
1335 return f"{self.name()}({self.fileDescriptor!r})"
1337 @property
1338 def fileDescriptor(self) -> FileDescriptor:
1339 """File descriptor associated with this formatter
1340 (`FileDescriptor`).
1341 """
1342 return self._fileDescriptor
1344 @property
1345 def dataId(self) -> DataCoordinate:
1346 """Return Data ID associated with this formatter (`DataCoordinate`)."""
1347 return self._dataId
1349 @property
1350 def writeParameters(self) -> Mapping[str, Any]:
1351 """Parameters to use when writing out datasets."""
1352 if self._writeParameters is not None:
1353 return self._writeParameters
1354 return {}
1356 @property
1357 def writeRecipes(self) -> Mapping[str, Any]:
1358 """Detailed write Recipes indexed by recipe name."""
1359 if self._writeRecipes is not None:
1360 return self._writeRecipes
1361 return {}
1363 def can_accept(self, in_memory_dataset: Any) -> bool:
1364 """Indicate whether this formatter can accept the specified
1365 storage class directly.
1367 Parameters
1368 ----------
1369 in_memory_dataset : `object`
1370 The dataset that is to be written.
1372 Returns
1373 -------
1374 accepts : `bool`
1375 If `True` the formatter can write data of this type without
1376 requiring datastore to convert it. If `False` the datastore
1377 will attempt to convert before writing.
1379 Notes
1380 -----
1381 The base class checks that the given python type matches
1382 the python type specified for this formatter when
1383 constructed.
1384 """
1385 return isinstance(in_memory_dataset, self.file_descriptor.storageClass.pytype)
1387 @classmethod
1388 def validateWriteRecipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
1389 """Validate supplied recipes for this formatter.
1391 The recipes are supplemented with default values where appropriate.
1393 Parameters
1394 ----------
1395 recipes : `dict`
1396 Recipes to validate.
1398 Returns
1399 -------
1400 validated : `dict`
1401 Validated recipes.
1403 Raises
1404 ------
1405 RuntimeError
1406 Raised if validation fails. The default implementation raises
1407 if any recipes are given.
1408 """
1409 if recipes:
1410 raise RuntimeError(f"This formatter does not understand these writeRecipes: {recipes}")
1411 return recipes
1413 @classmethod
1414 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
1415 return cls.validateWriteRecipes(recipes)
1417 @classmethod
1418 def name(cls) -> str:
1419 """Return the fully qualified name of the formatter.
1421 Returns
1422 -------
1423 name : `str`
1424 Fully-qualified name of formatter class.
1425 """
1426 return get_full_type_name(cls)
1428 @abstractmethod
1429 def read(self, component: str | None = None) -> Any:
1430 """Read a Dataset.
1432 Parameters
1433 ----------
1434 component : `str`, optional
1435 Component to read from the file. Only used if the `StorageClass`
1436 for reading differed from the `StorageClass` used to write the
1437 file.
1439 Returns
1440 -------
1441 inMemoryDataset : `object`
1442 The requested Dataset.
1443 """
1444 raise FormatterNotImplementedError("Type does not support reading")
1446 @abstractmethod
1447 def write(self, inMemoryDataset: Any) -> None:
1448 """Write a Dataset.
1450 Parameters
1451 ----------
1452 inMemoryDataset : `object`
1453 The Dataset to store.
1454 """
1455 raise FormatterNotImplementedError("Type does not support writing")
1457 @classmethod
1458 def can_read_bytes(cls) -> bool:
1459 """Indicate if this formatter can format from bytes.
1461 Returns
1462 -------
1463 can : `bool`
1464 `True` if the `fromBytes` method is implemented.
1465 """
1466 # We have no property to read so instead try to format from a byte
1467 # and see what happens
1468 try:
1469 # We know the arguments are incompatible
1470 cls.fromBytes(cls, b"") # type: ignore
1471 except FormatterNotImplementedError:
1472 return False
1473 except Exception:
1474 # There will be problems with the bytes we are supplying so ignore
1475 pass
1476 return True
1478 def fromBytes(self, serializedDataset: bytes, component: str | None = None) -> object:
1479 """Read serialized data into a Dataset or its component.
1481 Parameters
1482 ----------
1483 serializedDataset : `bytes`
1484 Bytes object to unserialize.
1485 component : `str`, optional
1486 Component to read from the Dataset. Only used if the `StorageClass`
1487 for reading differed from the `StorageClass` used to write the
1488 file.
1490 Returns
1491 -------
1492 inMemoryDataset : `object`
1493 The requested data as a Python object. The type of object
1494 is controlled by the specific formatter.
1495 """
1496 raise FormatterNotImplementedError("Type does not support reading from bytes.")
1498 def toBytes(self, inMemoryDataset: Any) -> bytes:
1499 """Serialize the Dataset to bytes based on formatter.
1501 Parameters
1502 ----------
1503 inMemoryDataset : `object`
1504 The Python object to serialize.
1506 Returns
1507 -------
1508 serializedDataset : `bytes`
1509 Bytes representing the serialized dataset.
1510 """
1511 raise FormatterNotImplementedError("Type does not support writing to bytes.")
1513 @contextlib.contextmanager
1514 def _updateLocation(self, location: Location | None) -> Iterator[Location]:
1515 """Temporarily replace the location associated with this formatter.
1517 Parameters
1518 ----------
1519 location : `Location`
1520 New location to use for this formatter. If `None` the
1521 formatter will not change but it will still return
1522 the old location. This allows it to be used in a code
1523 path where the location may not need to be updated
1524 but the with block is still convenient.
1526 Yields
1527 ------
1528 old : `Location`
1529 The old location that will be restored.
1531 Notes
1532 -----
1533 This is an internal method that should be used with care.
1534 It may change in the future. Should be used as a context
1535 manager to restore the location when the temporary is no
1536 longer required.
1537 """
1538 old = self._fileDescriptor.location
1539 try:
1540 if location is not None:
1541 self._fileDescriptor.location = location
1542 yield old
1543 finally:
1544 if location is not None:
1545 self._fileDescriptor.location = old
1547 def make_updated_location(self, location: Location) -> Location:
1548 """Return a new `Location` updated with this formatter's extension.
1550 Parameters
1551 ----------
1552 location : `Location`
1553 The location to update.
1555 Returns
1556 -------
1557 updated : `Location`
1558 A new `Location` with a new file extension applied.
1560 Raises
1561 ------
1562 NotImplementedError
1563 Raised if there is no ``extension`` attribute associated with
1564 this formatter.
1566 Notes
1567 -----
1568 This method is available to all Formatters but might not be
1569 implemented by all formatters. It requires that a formatter set
1570 an ``extension`` attribute containing the file extension used when
1571 writing files. If ``extension`` is `None` the supplied file will
1572 not be updated. Not all formatters write files so this is not
1573 defined in the base class.
1574 """
1575 location = location.clone()
1576 try:
1577 # We are deliberately allowing extension to be undefined by
1578 # default in the base class and mypy complains.
1579 location.updateExtension(self.extension) # type:ignore
1580 except AttributeError:
1581 raise NotImplementedError("No file extension registered with this formatter") from None
1582 return location
1584 @classmethod
1585 def validate_extension(cls, location: Location) -> None:
1586 """Check the extension of the provided location for compatibility.
1588 Parameters
1589 ----------
1590 location : `Location`
1591 Location from which to extract a file extension.
1593 Raises
1594 ------
1595 NotImplementedError
1596 Raised if file extensions are a concept not understood by this
1597 formatter.
1598 ValueError
1599 Raised if the formatter does not understand this extension.
1601 Notes
1602 -----
1603 This method is available to all Formatters but might not be
1604 implemented by all formatters. It requires that a formatter set
1605 an ``extension`` attribute containing the file extension used when
1606 writing files. If ``extension`` is `None` only the set of supported
1607 extensions will be examined.
1608 """
1609 supported = set(cls.supportedExtensions)
1611 try:
1612 # We are deliberately allowing extension to be undefined by
1613 # default in the base class and mypy complains.
1614 default = cls.extension # type: ignore
1615 except AttributeError:
1616 raise NotImplementedError("No file extension registered with this formatter") from None
1618 # If extension is implemented as an instance property it won't return
1619 # a string when called as a class property. Assume that
1620 # the supported extensions class property is complete.
1621 if default is not None and isinstance(default, str):
1622 supported.add(default)
1624 # Get the file name from the uri
1625 file = location.uri.basename()
1627 # Check that this file name ends with one of the supported extensions.
1628 # This is less prone to confusion than asking the location for
1629 # its extension and then doing a set comparison
1630 for ext in supported:
1631 if file.endswith(ext):
1632 return
1634 raise ValueError(
1635 f"Extension '{location.getExtension()}' on '{location}' "
1636 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})"
1637 )
1639 def predict_path(self) -> str:
1640 """Return the path that would be returned by write.
1642 Does not write any data file.
1644 Uses the `FileDescriptor` associated with the instance.
1646 Returns
1647 -------
1648 path : `str`
1649 Path within datastore that would be associated with the location
1650 stored in this `Formatter`.
1651 """
1652 updated = self.make_updated_location(self.fileDescriptor.location)
1653 return updated.pathInStore.path
1655 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
1656 """Segregate the supplied parameters.
1658 This splits the parameters into those understood by the
1659 formatter and those not understood by the formatter.
1661 Any unsupported parameters are assumed to be usable by associated
1662 assemblers.
1664 Parameters
1665 ----------
1666 parameters : `dict`, optional
1667 Parameters with values that have been supplied by the caller
1668 and which might be relevant for the formatter. If `None`
1669 parameters will be read from the registered `FileDescriptor`.
1671 Returns
1672 -------
1673 supported : `dict`
1674 Those parameters supported by this formatter.
1675 unsupported : `dict`
1676 Those parameters not supported by this formatter.
1677 """
1678 if parameters is None:
1679 parameters = self.fileDescriptor.parameters
1681 if parameters is None:
1682 return {}, {}
1684 if self.unsupportedParameters is None:
1685 # Support none of the parameters
1686 return {}, parameters.copy()
1688 # Start by assuming all are supported
1689 supported = parameters.copy()
1690 unsupported = {}
1692 # And remove any we know are not supported
1693 for p in set(supported):
1694 if p in self.unsupportedParameters:
1695 unsupported[p] = supported.pop(p)
1697 return supported, unsupported
1699 # Support classic V1 interface.
1700 makeUpdatedLocation = make_updated_location
1701 validateExtension = validate_extension
1702 segregateParameters = segregate_parameters
1703 predictPath = predict_path
1705 # Compatibility with V2 properties.
1706 @property
1707 def write_parameters(self) -> Mapping[str, Any]:
1708 return self.writeParameters
1710 @property
1711 def write_recipes(self) -> Mapping[str, Any]:
1712 return self.writeRecipes
1714 @property
1715 def file_descriptor(self) -> FileDescriptor:
1716 return self.fileDescriptor
1718 @property
1719 def data_id(self) -> DataCoordinate:
1720 return self.dataId
1723class FormatterFactory:
1724 """Factory for `Formatter` instances."""
1726 defaultKey = LookupKey("default")
1727 """Configuration key associated with default write parameter settings."""
1729 writeRecipesKey = LookupKey("write_recipes")
1730 """Configuration key associated with write recipes."""
1732 def __init__(self) -> None:
1733 self._mappingFactory = MappingFactory(Formatter)
1735 def __contains__(self, key: LookupKey | str) -> bool:
1736 """Indicate whether the supplied key is present in the factory.
1738 Parameters
1739 ----------
1740 key : `LookupKey`, `str` or objects with ``name`` attribute
1741 Key to use to lookup in the factory whether a corresponding
1742 formatter is present.
1744 Returns
1745 -------
1746 in : `bool`
1747 `True` if the supplied key is present in the factory.
1748 """
1749 return key in self._mappingFactory
1751 def registerFormatters(self, config: Config, *, universe: DimensionUniverse) -> None:
1752 """Bulk register formatters from a config.
1754 Parameters
1755 ----------
1756 config : `Config`
1757 ``formatters`` section of a configuration.
1758 universe : `DimensionUniverse`, optional
1759 Set of all known dimensions, used to expand and validate any used
1760 in lookup keys.
1762 Notes
1763 -----
1764 The configuration can include one level of hierarchy where an
1765 instrument-specific section can be defined to override more general
1766 template specifications. This is represented in YAML using a
1767 key of form ``instrument<name>`` which can then define templates
1768 that will be returned if a `DatasetRef` contains a matching instrument
1769 name in the data ID.
1771 The config is parsed using the function
1772 `~lsst.daf.butler.configSubset.processLookupConfigs`.
1774 The values for formatter entries can be either a simple string
1775 referring to a python type or a dict representing the formatter and
1776 parameters to be hard-coded into the formatter constructor. For
1777 the dict case the following keys are supported:
1779 - formatter: The python type to be used as the formatter class.
1780 - parameters: A further dict to be passed directly to the
1781 ``write_parameters`` Formatter constructor to seed it.
1782 These parameters are validated at instance creation and not at
1783 configuration.
1785 Additionally, a special ``default`` section can be defined that
1786 uses the formatter type (class) name as the keys and specifies
1787 default write parameters that should be used whenever an instance
1788 of that class is constructed.
1790 .. code-block:: yaml
1792 formatters:
1793 default:
1794 lsst.daf.butler.formatters.example.ExampleFormatter:
1795 max: 10
1796 min: 2
1797 comment: Default comment
1798 calexp: lsst.daf.butler.formatters.example.ExampleFormatter
1799 coadd:
1800 formatter: lsst.daf.butler.formatters.example.ExampleFormatter
1801 parameters:
1802 max: 5
1804 Any time an ``ExampleFormatter`` is constructed it will use those
1805 parameters. If an explicit entry later in the configuration specifies
1806 a different set of parameters, the two will be merged with the later
1807 entry taking priority. In the example above ``calexp`` will use
1808 the default parameters but ``coadd`` will override the value for
1809 ``max``.
1811 Formatter configuration can also include a special section describing
1812 collections of write parameters that can be accessed through a
1813 simple label. This allows common collections of options to be
1814 specified in one place in the configuration and reused later.
1815 The ``write_recipes`` section is indexed by Formatter class name
1816 and each key is the label to associate with the parameters.
1818 .. code-block:: yaml
1820 formatters:
1821 write_recipes:
1822 lsst.obs.base.formatters.fitsExposure.FixExposureFormatter:
1823 lossless:
1824 ...
1825 noCompression:
1826 ...
1828 By convention a formatter that uses write recipes will support a
1829 ``recipe`` write parameter that will refer to a recipe name in
1830 the ``write_recipes`` component. The `Formatter` will be constructed
1831 in the `FormatterFactory` with all the relevant recipes and
1832 will not attempt to filter by looking at ``write_parameters`` in
1833 advance. See the specific formatter documentation for details on
1834 acceptable recipe options.
1835 """
1836 allowed_keys = {"formatter", "parameters"}
1838 contents = processLookupConfigs(config, allow_hierarchy=True, universe=universe)
1840 # Extract any default parameter settings
1841 defaultParameters = contents.get(self.defaultKey, {})
1842 if not isinstance(defaultParameters, Mapping):
1843 raise RuntimeError(
1844 "Default formatter parameters in config can not be a single string"
1845 f" (got: {type(defaultParameters)})"
1846 )
1848 # Extract any global write recipes -- these are indexed by
1849 # Formatter class name.
1850 writeRecipes = contents.get(self.writeRecipesKey, {})
1851 if isinstance(writeRecipes, str):
1852 raise RuntimeError(
1853 f"The formatters.{self.writeRecipesKey} section must refer to a dict not '{writeRecipes}'"
1854 )
1856 for key, f in contents.items():
1857 # default is handled in a special way
1858 if key == self.defaultKey:
1859 continue
1860 if key == self.writeRecipesKey:
1861 continue
1863 # Can be a str or a dict.
1864 specificWriteParameters = {}
1865 if isinstance(f, str):
1866 formatter = f
1867 elif isinstance(f, Mapping):
1868 all_keys = set(f)
1869 unexpected_keys = all_keys - allowed_keys
1870 if unexpected_keys:
1871 raise ValueError(f"Formatter {key} uses unexpected keys {unexpected_keys} in config")
1872 if "formatter" not in f:
1873 raise ValueError(f"Mandatory 'formatter' key missing for formatter key {key}")
1874 formatter = f["formatter"]
1875 if "parameters" in f:
1876 specificWriteParameters = f["parameters"]
1877 else:
1878 raise ValueError(f"Formatter for key {key} has unexpected value: '{f}'")
1880 # Apply any default parameters for this formatter
1881 writeParameters = copy.deepcopy(defaultParameters.get(formatter, {}))
1882 writeParameters.update(specificWriteParameters)
1884 kwargs: dict[str, Any] = {}
1885 if writeParameters:
1886 kwargs["write_parameters"] = writeParameters
1888 if formatter in writeRecipes:
1889 kwargs["write_recipes"] = writeRecipes[formatter]
1891 self.registerFormatter(key, formatter, **kwargs)
1893 def getLookupKeys(self) -> set[LookupKey]:
1894 """Retrieve the look up keys for all the registry entries.
1896 Returns
1897 -------
1898 keys : `set` of `LookupKey`
1899 The keys available for matching in the registry.
1900 """
1901 return self._mappingFactory.getLookupKeys()
1903 def getFormatterClassWithMatch(
1904 self, entity: Entity
1905 ) -> tuple[LookupKey, type[Formatter | FormatterV2], dict[str, Any]]:
1906 """Get the matching formatter class along with the registry key.
1908 Parameters
1909 ----------
1910 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1911 Entity to use to determine the formatter to return.
1912 `StorageClass` will be used as a last resort if `DatasetRef`
1913 or `DatasetType` instance is provided. Supports instrument
1914 override if a `DatasetRef` is provided configured with an
1915 ``instrument`` value for the data ID.
1917 Returns
1918 -------
1919 matchKey : `LookupKey`
1920 The key that resulted in the successful match.
1921 formatter : `type`
1922 The class of the registered formatter.
1923 formatter_kwargs : `dict`
1924 Keyword arguments that are associated with this formatter entry.
1925 """
1926 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
1927 matchKey, formatter, formatter_kwargs = self._mappingFactory.getClassFromRegistryWithMatch(names)
1928 log.debug(
1929 "Retrieved formatter %s from key '%s' for entity '%s'",
1930 get_full_type_name(formatter),
1931 matchKey,
1932 entity,
1933 )
1935 return matchKey, formatter, formatter_kwargs
1937 def getFormatterClass(self, entity: Entity) -> type:
1938 """Get the matching formatter class.
1940 Parameters
1941 ----------
1942 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1943 Entity to use to determine the formatter to return.
1944 `StorageClass` will be used as a last resort if `DatasetRef`
1945 or `DatasetType` instance is provided. Supports instrument
1946 override if a `DatasetRef` is provided configured with an
1947 ``instrument`` value for the data ID.
1949 Returns
1950 -------
1951 formatter : `type`
1952 The class of the registered formatter.
1953 """
1954 _, formatter, _ = self.getFormatterClassWithMatch(entity)
1955 return formatter
1957 def getFormatterWithMatch(
1958 self, entity: Entity, *args: Any, **kwargs: Any
1959 ) -> tuple[LookupKey, Formatter | FormatterV2]:
1960 """Get a new formatter instance along with the matching registry key.
1962 Parameters
1963 ----------
1964 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1965 Entity to use to determine the formatter to return.
1966 `StorageClass` will be used as a last resort if `DatasetRef`
1967 or `DatasetType` instance is provided. Supports instrument
1968 override if a `DatasetRef` is provided configured with an
1969 ``instrument`` value for the data ID.
1970 *args : `tuple`
1971 Positional arguments to use pass to the object constructor.
1972 **kwargs
1973 Keyword arguments to pass to object constructor.
1975 Returns
1976 -------
1977 matchKey : `LookupKey`
1978 The key that resulted in the successful match.
1979 formatter : `Formatter`
1980 An instance of the registered formatter.
1981 """
1982 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
1983 matchKey, formatter = self._mappingFactory.getFromRegistryWithMatch(names, *args, **kwargs)
1984 log.debug(
1985 "Retrieved formatter %s from key '%s' for entity '%s'",
1986 get_full_type_name(formatter),
1987 matchKey,
1988 entity,
1989 )
1991 return matchKey, formatter
1993 def getFormatter(self, entity: Entity, *args: Any, **kwargs: Any) -> Formatter | FormatterV2:
1994 """Get a new formatter instance.
1996 Parameters
1997 ----------
1998 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1999 Entity to use to determine the formatter to return.
2000 `StorageClass` will be used as a last resort if `DatasetRef`
2001 or `DatasetType` instance is provided. Supports instrument
2002 override if a `DatasetRef` is provided configured with an
2003 ``instrument`` value for the data ID.
2004 *args : `tuple`
2005 Positional arguments to use pass to the object constructor.
2006 **kwargs
2007 Keyword arguments to pass to object constructor.
2009 Returns
2010 -------
2011 formatter : `Formatter`
2012 An instance of the registered formatter.
2013 """
2014 _, formatter = self.getFormatterWithMatch(entity, *args, **kwargs)
2015 return formatter
2017 def registerFormatter(
2018 self,
2019 type_: LookupKey | str | StorageClass | DatasetType,
2020 formatter: str,
2021 *,
2022 overwrite: bool = False,
2023 **kwargs: Any,
2024 ) -> None:
2025 """Register a `Formatter`.
2027 Parameters
2028 ----------
2029 type_ : `LookupKey`, `str`, `StorageClass` or `DatasetType`
2030 Type for which this formatter is to be used. If a `LookupKey`
2031 is not provided, one will be constructed from the supplied string
2032 or by using the ``name`` property of the supplied entity.
2033 formatter : `str` or class of type `Formatter`
2034 Identifies a `Formatter` subclass to use for reading and writing
2035 Datasets of this type. Can be a `Formatter` class.
2036 overwrite : `bool`, optional
2037 If `True` an existing entry will be replaced by the new value.
2038 Default is `False`.
2039 **kwargs
2040 Keyword arguments to always pass to object constructor when
2041 retrieved.
2043 Raises
2044 ------
2045 ValueError
2046 Raised if the formatter does not name a valid formatter type and
2047 ``overwrite`` is `False`.
2048 """
2049 self._mappingFactory.placeInRegistry(type_, formatter, overwrite=overwrite, **kwargs)
2052class FormatterV1inV2(FormatterV2):
2053 """An implementation of a V2 formatter that provides a compatibility
2054 interface for V1 formatters.
2056 Parameters
2057 ----------
2058 file_descriptor : `FileDescriptor`, optional
2059 Identifies the file to read or write, and the associated storage
2060 classes and parameter information. Its value can be `None` if the
2061 caller will never call `Formatter.read` or `Formatter.write`.
2062 ref : `DatasetRef`
2063 The dataset associated with this formatter. Should not be a component
2064 dataset ref.
2065 formatter : `Formatter`
2066 A version 1 `Formatter` instance. The V2 formatter layer forwards calls
2067 to this formatter.
2068 write_parameters : `dict`, optional
2069 Any parameters to be hard-coded into this instance to control how
2070 the dataset is serialized.
2071 write_recipes : `dict`, optional
2072 Detailed write Recipes indexed by recipe name.
2073 **kwargs
2074 Additional arguments that will be ignored but allow for
2075 `Formatter` V1 parameters to be given.
2076 """
2078 can_read_from_local_file = True
2079 """This formatter can read from a local file."""
2081 def __init__(
2082 self,
2083 file_descriptor: FileDescriptor,
2084 *,
2085 ref: DatasetRef,
2086 formatter: Formatter,
2087 write_parameters: Mapping[str, Any] | None = None,
2088 write_recipes: Mapping[str, Any] | None = None,
2089 # Compatibility parameters. Unused in v2.
2090 **kwargs: Any,
2091 ):
2092 if not isinstance(formatter, Formatter):
2093 raise TypeError(f"Formatter parameter was not a V1 formatter (was {type(formatter)})")
2095 # Replace the class property with instance values from this
2096 # V1 formatter so that the V2 __init__ will be able to validate.
2097 self.supported_write_parameters = formatter.supportedWriteParameters # type: ignore
2098 self._formatter = formatter
2100 super().__init__(
2101 file_descriptor, ref=ref, write_parameters=write_parameters, write_recipes=write_recipes
2102 )
2104 def get_write_extension(self) -> str:
2105 ext = self._formatter.extension
2106 return ext if ext is not None else ""
2108 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
2109 return self._formatter.segregate_parameters(parameters)
2111 def validate_write_recipes( # type: ignore
2112 self,
2113 recipes: Mapping[str, Any] | None,
2114 ) -> Mapping[str, Any] | None:
2115 # This should be a class method but a class method can not work
2116 # for a dynamic shim. Luckily the shim is only used as an instance.
2117 return self._formatter.validate_write_recipes(recipes)
2119 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any:
2120 # Need to temporarily override the location since the V1 formatter
2121 # will not know anything about this local file.
2123 # V2 does not have a fromBytes equivalent.
2124 if self._formatter.can_read_bytes():
2125 with open(path, "rb") as fd:
2126 serialized_dataset = fd.read()
2127 return self._formatter.fromBytes(serialized_dataset, component=component)
2129 location = Location(None, path)
2130 with self._formatter._updateLocation(location):
2131 try:
2132 result = self._formatter.read(component=component)
2133 except NotImplementedError:
2134 # V1 raises NotImplementedError but V2 is expecting something
2135 # slightly different.
2136 return NotImplemented
2137 return result
2139 def to_bytes(self, in_memory_dataset: Any) -> bytes:
2140 try:
2141 return self._formatter.toBytes(in_memory_dataset)
2142 except NotImplementedError as e:
2143 # V1 raises NotImplementedError but V2 is expecting something
2144 # slightly different.
2145 raise FormatterNotImplementedError(str(e)) from e
2147 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
2148 with self._formatter._updateLocation(Location(None, uri)):
2149 try:
2150 self._formatter.write(in_memory_dataset)
2151 except NotImplementedError as e:
2152 # V1 raises NotImplementedError but V2 is expecting something
2153 # slightly different.
2154 raise FormatterNotImplementedError(str(e)) from e
2157# Type to use when allowing a Formatter or its class name
2158FormatterParameter: TypeAlias = str | type[Formatter] | Formatter | FormatterV2 | type[FormatterV2]