Coverage for python/lsst/images/json/_input_archive.py: 70%
70 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-08 09:10 +0000
1# This file is part of lsst-images.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12from __future__ import annotations
14__all__ = ("JsonInputArchive",)
16from collections.abc import Callable, Iterator
17from contextlib import contextmanager
18from types import EllipsisType
19from typing import IO, TYPE_CHECKING, Any, Self
21import astropy.table
22import numpy as np
23from pydantic_core import from_json
25from lsst.resources import ResourcePath, ResourcePathExpression
27from .._transforms import FrameSet
28from ..serialization import (
29 ArchiveInfo,
30 ArchiveReadError,
31 ArchiveTree,
32 ArrayReferenceModel,
33 InlineArrayModel,
34 InputArchive,
35 JsonRef,
36 TableModel,
37 no_header_updates,
38 parameterize_tree,
39 tree_class_for_info,
40)
41from ..serialization._backends import _is_binary_stream
43if TYPE_CHECKING:
44 import astropy.io.fits
47class JsonInputArchive(InputArchive[JsonRef]):
48 """An implementation of the `.serialization.InputArchive` interface that
49 reads from JSON files.
51 Parameters
52 ----------
53 indirect
54 The `.serialization.ArchiveTree.indirect` attribute of the root
55 serialization model.
56 """
58 @classmethod
59 def get_basic_info(cls, path: ResourcePathExpression) -> ArchiveInfo:
60 """Read the top-level tree's ``schema_url``; JSON has no container
61 format version.
63 This parses the whole document. Unlike the FITS and NDF backends
64 there is no cheap header to read: ``schema_url`` is a computed field
65 serialized after the (potentially large) ``indirect`` payload, and
66 nested trees carry their own ``schema_url``, so a bounded prefix
67 cannot identify the top-level tree reliably. JSON is not intended
68 for large pixel archives, where FITS or NDF should be used instead.
70 Parameters
71 ----------
72 path
73 Path to the archive to read.
74 """
75 raw = from_json(ResourcePath(path).read())
76 if not isinstance(raw, dict) or not raw.get("schema_url"): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true
77 raise ArchiveReadError(f"{path!r} has no schema_url in its top-level JSON tree.")
78 return ArchiveInfo.from_schema_url(raw["schema_url"], format_version=None)
80 @classmethod
81 @contextmanager
82 def open_tree(
83 cls,
84 path: ResourcePathExpression | IO[bytes],
85 *,
86 partial: bool = True,
87 **backend_kwargs: Any,
88 ) -> Iterator[tuple[Self, ArchiveTree, ArchiveInfo]]:
89 """Parse the JSON tree and yield ``(archive, tree, info)``.
91 Parameters
92 ----------
93 path
94 File resource to open, or a seekable binary stream containing
95 the file's content.
96 partial
97 Ignored. The entire JSON file is always read into memory.
98 **backend_kwargs
99 No keyword parameters are supported by this backend.
100 """
101 if _is_binary_stream(path):
102 raw = path.read()
103 else:
104 raw = ResourcePath(path).read()
105 parsed = from_json(raw)
106 if not isinstance(parsed, dict) or not parsed.get("schema_url"): 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 raise ArchiveReadError(f"{path!r} has no schema_url in its top-level JSON tree.")
108 info = ArchiveInfo.from_schema_url(parsed["schema_url"], format_version=None)
109 tree_cls = tree_class_for_info(info, path)
110 parameterized = parameterize_tree(tree_cls, JsonRef)
111 tree = parameterized.model_validate_json(raw)
112 archive = cls(tree.indirect)
113 try:
114 yield archive, tree, info
115 finally:
116 tree.indirect = []
118 def __init__(self, indirect: list[Any] | None = None) -> None:
119 self._indirect = indirect if indirect is not None else []
120 self._deserialized_pointer_cache: dict[int, Any] = {}
122 def deserialize_pointer[U: ArchiveTree, V](
123 self,
124 pointer: JsonRef,
125 model_type: type[U],
126 deserializer: Callable[[U, InputArchive[JsonRef]], V],
127 ) -> V:
128 index = int(pointer.ref.removeprefix("#/indirect/"))
129 if (existing := self._deserialized_pointer_cache.get(index)) is not None:
130 return existing
131 model = model_type.model_validate(self._indirect[index])
132 result = deserializer(model, self)
133 self._deserialized_pointer_cache[index] = result
134 return result
136 def get_frame_set(self, ref: JsonRef) -> FrameSet:
137 index = int(ref.ref.removeprefix("#/indirect/"))
138 try:
139 result = self._deserialized_pointer_cache[index]
140 except KeyError:
141 raise AssertionError(
142 f"Frame set at {ref.model_dump_json(indent=2)} must be deserialized "
143 "before any dependent transform can be."
144 ) from None
145 if not isinstance(result, FrameSet):
146 raise ArchiveReadError(f"Expected a FrameSet instance at {ref.model_dump_json(indent=2)}.")
147 return result
149 def get_array(
150 self,
151 model: ArrayReferenceModel | InlineArrayModel,
152 *,
153 slices: tuple[slice, ...] | EllipsisType = ...,
154 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
155 ) -> np.ndarray:
156 if not isinstance(model, InlineArrayModel): 156 ↛ 157line 156 didn't jump to line 157 because the condition on line 156 was never true
157 raise ArchiveReadError("Only inline arrays are supported in JSON archives.")
158 return np.array(model.data, dtype=model.datatype.to_numpy())[slices]
160 def get_table(
161 self,
162 model: TableModel,
163 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
164 ) -> astropy.table.Table:
165 result = astropy.table.Table(meta=model.meta)
166 for column_model in model.columns:
167 if not isinstance(column_model.data, InlineArrayModel): 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true
168 raise ArchiveReadError("Only inline arrays are supported in JSON archives.")
169 result[column_model.name] = astropy.table.Column(
170 column_model.data.data,
171 name=column_model.name,
172 dtype=column_model.data.datatype.to_numpy(),
173 unit=column_model.unit,
174 description=column_model.description,
175 meta=column_model.meta,
176 )
177 return result
179 def get_structured_array(
180 self,
181 model: TableModel,
182 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
183 ) -> np.ndarray:
184 table = self.get_table(model)
185 return table.as_array()