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