Coverage for python/lsst/images/cells/_provenance.py: 55%
264 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-10 01:39 -0800
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__ = ("CoaddProvenance", "CoaddProvenanceSerializationModel")
16from collections.abc import Iterable
17from typing import TYPE_CHECKING, Any, ClassVar
19import astropy.table
20import astropy.units as u
21import numpy as np
22import pydantic
24from .._cell_grid import CellGridBounds, CellIJ
25from .._geom import XY, Box, NoOverlapError
26from .._polygon import Polygon, Region, RegionDomainError
27from .._transforms import GeneralFrame, SkyProjection
28from ..describe import DescribableMixin, DescribeOptions, FieldRole, Report, ReportField
29from ..serialization import ArchiveTree, InputArchive, InvalidParameterError, OutputArchive, TableModel
31if TYPE_CHECKING:
32 try:
33 from lsst.afw.geom import Polygon as LegacyPolygon
34 from lsst.afw.image import CoaddInputs as LegacyCoaddInputs
35 from lsst.cell_coadds import CoaddInputs as LegacyCellCoaddInputs
36 from lsst.cell_coadds import MultipleCellCoadd as LegacyMultipleCellCoadd
37 from lsst.cell_coadds import ObservationIdentifiers as LegacyObservationIdentifiers
38 from lsst.skymap import Index2D as LegacyIndex2D
39 except ImportError:
40 type LegacyIndex2D = Any # type: ignore[no-redef]
41 type LegacyCellCoaddInputs = Any # type: ignore[no-redef]
42 type LegacyCoaddInputs = Any # type: ignore[no-redef]
43 type LegacyPolygon = Any # type: ignore[no-redef]
44 type LegacyMultipleCellCoadd = Any # type: ignore[no-redef]
45 type LegacyObservationIdentifiers = Any # type: ignore[no-redef]
48class CoaddProvenance(DescribableMixin):
49 """A pair of tables that record the inputs to a cell-based coadd.
51 Parameters
52 ----------
53 inputs
54 A table of {visit, detector} combinations that contribute to any cell
55 in the coadd.
56 contributions
57 A table of {visit, detector, cell} combinations that describe how an
58 observation contributed to a cell.
60 Raises
61 ------
62 ValueError
63 Raised if ``inputs`` has no rows; provenance that records no input
64 image is not meaningful.
66 Notes
67 -----
68 This object can represent the provenance of a whole patch, a single cell,
69 or anything in between. In the single-cell case, the ``inputs`` and
70 ``contributions`` tables have the same number of rows (but may not be
71 ordered the same way!).
72 """
74 def __init__(self, inputs: astropy.table.Table, contributions: astropy.table.Table) -> None:
75 if not len(inputs):
76 raise ValueError("Coadd provenance must contain at least one input image.")
77 self._inputs = inputs
78 self._contributions = contributions
80 _INPUT_TABLE_COLUMNS: ClassVar[list[tuple[str, type, str]]] = [
81 ("instrument", np.object_, "Name of the instrument."),
82 ("visit", np.uint64, "ID of the visit."),
83 ("detector", np.uint16, "ID of the detector."),
84 ("physical_filter", np.object_, "Full name of the bandpass filter."),
85 ("day_obs", np.uint32, "Observation night as a YYYYMMDD integer."),
86 (
87 "polygon",
88 np.object_,
89 (
90 "Polygon that approximates the overlap of the observation and the coadd patch, "
91 "in coadd coordinates."
92 ),
93 ),
94 ]
96 _CONTRIBUTION_TABLE_COLUMNS: ClassVar[list[tuple[str, type, str, u.UnitBase | None]]] = [
97 ("cell_i", np.uint16, "Y-axis index of the cell within the patch.", None),
98 ("cell_j", np.uint16, "X-axis index of the cell within the patch.", None),
99 ("instrument", np.object_, "Name of the instrument.", None),
100 ("visit", np.uint64, "ID of the visit.", None),
101 ("detector", np.uint16, "ID of the detector.", None),
102 ("overlaps_center", np.bool_, "Whether a this observation overlaps the center of the cell.", None),
103 ("overlap_fraction", np.float64, "Fraction of the cell that is covered by the overlap region.", None),
104 ("unmasked_fraction", np.float64, "Fraction of the cell propagated to the coadd.", None),
105 ("weight", np.float64, "Weight to be used for this input in this cell.", None),
106 ("psf_shape_xx", np.float64, "Second order moments of the PSF.", u.pix**2),
107 ("psf_shape_yy", np.float64, "Second order moments of the PSF.", u.pix**2),
108 ("psf_shape_xy", np.float64, "Second order moments of the PSF.", u.pix**2),
109 (
110 "psf_shape_flag",
111 np.bool_,
112 "Flag indicating whether the PSF shape measurement was successful.",
113 None,
114 ),
115 ]
117 @classmethod
118 def make_empty_input_table(cls, n_rows: int) -> astropy.table.Table:
119 """Make an empty `inputs` table with a set number of rows.
121 Parameters
122 ----------
123 n_rows
124 Number of rows in the new table.
125 """
126 return astropy.table.Table(
127 [
128 astropy.table.Column(name=name, length=n_rows, dtype=dtype, description=description)
129 for name, dtype, description in cls._INPUT_TABLE_COLUMNS
130 ]
131 )
133 @classmethod
134 def make_empty_contribution_table(cls, n_rows: int) -> astropy.table.Table:
135 """Make an empty `contributions` table with a set number of rows.
137 Parameters
138 ----------
139 n_rows
140 Number of rows in the new table.
141 """
142 return astropy.table.Table(
143 [
144 astropy.table.Column(
145 name=name, length=n_rows, dtype=dtype, description=description, unit=unit
146 )
147 for name, dtype, description, unit in cls._CONTRIBUTION_TABLE_COLUMNS
148 ]
149 )
151 @property
152 def inputs(self) -> astropy.table.Table:
153 """A table of {visit, detector} combinations that contribute to any
154 cell in the coadd.
155 """
156 return self._inputs
158 @property
159 def contributions(self) -> astropy.table.Table:
160 """A table of {visit, detector, cell} combinations that describe how an
161 observation contributed to a cell.
162 """
163 return self._contributions
165 def __getitem__(self, cell: CellIJ) -> CoaddProvenance | None:
166 return self.subset([cell])
168 def subset(self, cells: Iterable[CellIJ]) -> CoaddProvenance | None:
169 """Return a new provenance object with just the given cells.
171 Parameters
172 ----------
173 cells
174 Cells to keep in the returned provenance.
176 Returns
177 -------
178 subset : `CoaddProvenance` or `None`
179 Provenance for ``cells`` only, or `None` if none of them have
180 contributions.
181 """
182 cells_to_keep = astropy.table.Table(
183 rows=[(index.i, index.j) for index in cells],
184 names=["cell_i", "cell_j"],
185 dtype=[np.uint16, np.uint16],
186 )
187 if not len(cells_to_keep): 187 ↛ 189line 187 didn't jump to line 189 because the condition on line 187 was never true
188 # astropy.table.join rejects an empty operand outright.
189 return None
190 contributions = astropy.table.join(self._contributions, cells_to_keep)
191 assert contributions.columns.keys() == {name for name, _, _, _ in self._CONTRIBUTION_TABLE_COLUMNS}
192 if not len(contributions):
193 # No inputs to derive either, and astropy.table.join rejects an
194 # empty operand outright.
195 return None
196 inputs = astropy.table.join(contributions["instrument", "visit", "detector"], self._inputs)
197 assert inputs.columns.keys() == {name for name, _, _ in self._INPUT_TABLE_COLUMNS}
198 return CoaddProvenance(inputs=inputs, contributions=contributions)
200 def _describe(
201 self,
202 options: DescribeOptions = DescribeOptions(),
203 /,
204 *,
205 bounds: CellGridBounds | None = None,
206 ) -> Report:
207 """Return a `Report` describing this provenance.
209 Parameters
210 ----------
211 options : `DescribeOptions`, optional
212 Rendering options. `DescribeOptions.brief` reports only the
213 number of input images, which no column scan is needed to count.
214 bounds : `CellGridBounds`, optional
215 Cells the image this provenance belongs to has data for. When
216 given, the number of cells with contributions is reported as a
217 fraction of them.
219 Notes
220 -----
221 The report summarizes the tables rather than rendering them. Rich
222 renders an embedded `astropy.table.Table` as plain text and never
223 consults the table's own ``_repr_html_``, so `inputs` and
224 `contributions` describe themselves better than this report could,
225 and a patch has tens of thousands of contribution rows.
226 """
227 n_inputs = len(self._inputs)
228 summary = f"CoaddProvenance({n_inputs} input image{'s' if n_inputs != 1 else ''})"
229 if options.brief:
230 return Report(type_name="CoaddProvenance", summary=summary)
231 fields: list[ReportField] = []
232 for column in ("instrument", "physical_filter"):
233 fields.append(
234 ReportField(
235 label=column,
236 value=", ".join(sorted({str(value) for value in self._inputs[column]})),
237 role=FieldRole.DERIVED,
238 )
239 )
240 n_visits = len(np.unique(self._inputs["visit"]))
241 input_images = f"{n_inputs} from {n_visits} visit{'s' if n_visits != 1 else ''}"
242 fields.append(ReportField(label="input images", value=input_images, role=FieldRole.DERIVED))
243 first_night = int(self._inputs["day_obs"].min())
244 last_night = int(self._inputs["day_obs"].max())
245 fields.append(
246 ReportField(
247 label="day_obs",
248 value=str(first_night) if first_night == last_night else f"{first_night} - {last_night}",
249 role=FieldRole.DERIVED,
250 )
251 )
252 _, counts = np.unique(
253 np.column_stack([self._contributions["cell_i"], self._contributions["cell_j"]]),
254 axis=0,
255 return_counts=True,
256 )
257 if bounds is not None:
258 n_cells_with_data = bounds.subgrid_size.i * bounds.subgrid_size.j - len(bounds.missing)
259 cells = f"{len(counts)} of {n_cells_with_data} with contributions"
260 else:
261 cells = f"{len(counts)} with contributions" if len(counts) else "none"
262 fields.append(ReportField(label="cells", value=cells, role=FieldRole.DERIVED))
263 if len(counts):
264 low = int(counts.min())
265 high = int(counts.max())
266 per_cell = (
267 f"{low} input image{'s' if low != 1 else ''}"
268 if low == high
269 else f"{low} - {high} input images (median {np.median(counts):g})"
270 )
271 fields.append(ReportField(label="per cell", value=per_cell, role=FieldRole.DERIVED))
272 return Report(type_name="CoaddProvenance", summary=summary, fields=fields)
274 def serialize(self, archive: OutputArchive[Any]) -> CoaddProvenanceSerializationModel:
275 """Serialize the provenance to an output archive.
277 Parameters
278 ----------
279 archive
280 Archive to write to.
281 """
282 # The tables are exposed and mutable, so a caller can empty one after
283 # construction. Catch that here rather than deep in the column
284 # rewriting below, where it surfaces as "max() iterable argument is
285 # empty".
286 assert len(self._inputs), "Coadd provenance must contain at least one input image."
287 inputs = self._inputs.copy(copy_data=False)
288 contributions = self._contributions.copy(copy_data=False)
289 instrument = CoaddProvenanceSerializationModel._fix_str_for_serialization(
290 "instrument", inputs, contributions
291 )
292 physical_filter = CoaddProvenanceSerializationModel._fix_str_for_serialization(
293 "physical_filter", inputs
294 )
295 CoaddProvenanceSerializationModel._fix_polygon_for_serialization(inputs)
296 inputs_model = archive.add_table(inputs, name="inputs")
297 contributions_model = archive.add_table(contributions, name="contributions")
298 return CoaddProvenanceSerializationModel(
299 instrument=instrument,
300 physical_filter=physical_filter,
301 inputs=inputs_model,
302 contributions=contributions_model,
303 )
305 @staticmethod
306 def from_legacy(legacy_cell_coadd: LegacyMultipleCellCoadd) -> CoaddProvenance:
307 """Extract provenance from a legacy
308 `lsst.cell_coadds.MultipleCellCoadd` object.
310 Parameters
311 ----------
312 legacy_cell_coadd
313 Legacy cell coadd to extract provenance from.
314 """
315 inputs = CoaddProvenance.make_empty_input_table(len(legacy_cell_coadd.common.visit_polygons))
316 for n, (legacy_identifiers, legacy_polygon) in enumerate(
317 legacy_cell_coadd.common.visit_polygons.items()
318 ):
319 inputs["instrument"][n] = legacy_identifiers.instrument
320 inputs["visit"][n] = legacy_identifiers.visit
321 inputs["detector"][n] = legacy_identifiers.detector
322 inputs["physical_filter"][n] = legacy_identifiers.physical_filter
323 inputs["day_obs"][n] = legacy_identifiers.day_obs
324 inputs["polygon"][n] = Polygon.from_legacy(legacy_polygon)
325 n_contributions = 0
326 for legacy_cell in legacy_cell_coadd.cells.values():
327 n_contributions += len(legacy_cell.inputs)
328 contributions = CoaddProvenance.make_empty_contribution_table(n_contributions)
329 n = 0
330 for legacy_cell in legacy_cell_coadd.cells.values():
331 for legacy_identifiers, legacy_inputs in legacy_cell.inputs.items():
332 contributions["cell_i"][n] = legacy_cell.identifiers.cell.y
333 contributions["cell_j"][n] = legacy_cell.identifiers.cell.x
334 contributions["instrument"][n] = legacy_identifiers.instrument
335 contributions["visit"][n] = legacy_identifiers.visit
336 contributions["detector"][n] = legacy_identifiers.detector
337 contributions["overlaps_center"][n] = legacy_inputs.overlaps_center
338 contributions["overlap_fraction"][n] = legacy_inputs.overlap_fraction
339 contributions["unmasked_fraction"][n] = legacy_inputs.unmasked_overlap_fraction
340 contributions["weight"][n] = legacy_inputs.weight
341 contributions["psf_shape_xx"][n] = legacy_inputs.psf_shape.getIxx()
342 contributions["psf_shape_yy"][n] = legacy_inputs.psf_shape.getIyy()
343 contributions["psf_shape_xy"][n] = legacy_inputs.psf_shape.getIxy()
344 contributions["psf_shape_flag"][n] = legacy_inputs.psf_shape_flag
345 n += 1
346 return CoaddProvenance(inputs=inputs, contributions=contributions)
348 def to_legacy_polygon_map(self) -> dict[LegacyObservationIdentifiers, LegacyPolygon]:
349 """Construct a legacy mapping from
350 `lsst.cell_coadds.ObservationIdentifiers` to `lsst.afw.geom.Polygon`
351 from the `inputs` table.
352 """
353 from lsst.cell_coadds import ObservationIdentifiers as LegacyObservationIdentifiers
355 return {
356 LegacyObservationIdentifiers(
357 instrument=str(row["instrument"]),
358 physical_filter=str(row["physical_filter"]),
359 visit=int(row["visit"]),
360 day_obs=int(row["day_obs"]),
361 detector=int(row["detector"]),
362 ): row["polygon"].to_legacy()
363 for row in self.inputs
364 }
366 def to_legacy_cell_coadd_inputs(
367 self, observations: Iterable[LegacyObservationIdentifiers] | None
368 ) -> dict[LegacyIndex2D, dict[LegacyObservationIdentifiers, LegacyCellCoaddInputs]]:
369 """Construct a mapping from legacy cell index to the list of legacy
370 input structs for that cell.
372 Parameters
373 ----------
374 observations
375 Observations to include, or `None` to include all observations
376 in the `inputs` table.
377 """
378 from lsst.afw.geom.ellipses import Quadrupole
379 from lsst.cell_coadds import CoaddInputs as LegacyCellCoaddInputs
380 from lsst.skymap import Index2D as LegacyIndex2D
382 if observations is None:
383 observations = self.to_legacy_polygon_map().keys()
384 observations_by_key: dict[tuple[str, int, int], LegacyObservationIdentifiers] = {
385 (obs.instrument, obs.visit, obs.detector): obs for obs in observations
386 }
387 result: dict[LegacyIndex2D, dict[LegacyObservationIdentifiers, LegacyCellCoaddInputs]] = {}
388 for row in self.contributions:
389 obs_key = (str(row["instrument"]), int(row["visit"]), int(row["detector"]))
390 obs = observations_by_key[obs_key]
391 cell_inputs = result.setdefault(LegacyIndex2D(x=int(row["cell_j"]), y=int(row["cell_i"])), {})
392 cell_inputs[obs] = LegacyCellCoaddInputs(
393 overlaps_center=bool(row["overlaps_center"]),
394 overlap_fraction=float(row["overlap_fraction"]),
395 unmasked_overlap_fraction=float(row["unmasked_fraction"]),
396 weight=float(row["weight"]),
397 psf_shape=Quadrupole(row["psf_shape_xx"], row["psf_shape_yy"], row["psf_shape_xy"]),
398 psf_shape_flag=bool(row["psf_shape_flag"]),
399 )
400 return result
402 @staticmethod
403 def from_legacy_lazy(
404 legacy_coadd_inputs: LegacyCoaddInputs,
405 bounds: CellGridBounds,
406 sky_projection: SkyProjection,
407 instrument: str,
408 ) -> CoaddProvenance:
409 """Convert provenance from a legacy
410 `lsst.afw.image.CoaddInputs` object.
412 Parameters
413 ----------
414 legacy_coadd_inputs
415 Legacy coadd-input provenance to convert.
416 bounds
417 The cell grid and known-missing cells.
418 sky_projection
419 The projection the coadd pixel grid to the sky.
420 instrument
421 Name of the instrument (for all inputs).
422 """
423 ccd_catalog = legacy_coadd_inputs.ccds
424 inputs = CoaddProvenance.make_empty_input_table(len(ccd_catalog))
425 inputs["instrument"] = instrument
426 contributions = CoaddProvenance.make_empty_contribution_table(bounds.n_cells * len(ccd_catalog))
427 contributions["instrument"] = instrument
428 contributions["unmasked_fraction"] = np.nan
429 m = 0
430 n = 0
431 for ccd_record in ccd_catalog:
432 visit_id = ccd_record.get("visit")
433 detector_id = ccd_record.get("ccd")
434 detector_region: Region = Polygon.from_box(Box.from_legacy(ccd_record.getBBox()))
435 if (legacy_polygon := ccd_record.getValidPolygon()) is not None:
436 detector_region = detector_region.intersection(Polygon.from_legacy(legacy_polygon))
437 detector_proj = SkyProjection.from_legacy(ccd_record.getWcs(), GeneralFrame(unit=u.pix))
438 detector_to_coadd = detector_proj.pixel_to_sky_transform.then(
439 sky_projection.sky_to_pixel_transform
440 )
441 try:
442 detector_coadd_region = detector_region.transform(detector_to_coadd)
443 except RegionDomainError:
444 # This input is outside the domain of the coadd's projection
445 # (e.g. on the far side of the sky from its tangent point),
446 # so it cannot overlap the coadd.
447 continue
448 try:
449 detector_patch_region = detector_coadd_region.intersection(bounds.bbox)
450 except NoOverlapError:
451 # This input does not overlap this coadd's bounds at all, as
452 # happens when converting from a larger (e.g. whole-tract
453 # template) CoaddInputs to a smaller (e.g. postage) coadd.
454 continue
455 inputs["visit"][n] = visit_id
456 inputs["detector"][n] = detector_id
457 inputs["physical_filter"][n] = ccd_record.get("filter")
458 inputs["day_obs"][n] = _extract_day_obs_from_visit(visit_id)
459 inputs["polygon"][n] = detector_patch_region
460 weight = ccd_record.get("weight")
461 for cell in bounds.cell_indices():
462 cell_box = bounds.grid.bbox_of(cell)
463 cell_center = XY(x=cell_box.x.center, y=cell_box.y.center)
464 try:
465 cell_region = detector_patch_region.intersection(cell_box)
466 except NoOverlapError:
467 continue
468 contributions["cell_i"][m] = cell.i
469 contributions["cell_j"][m] = cell.j
470 contributions["visit"][m] = visit_id
471 contributions["detector"][m] = detector_id
472 contributions["overlaps_center"][m] = cell_region.contains(cell_center)
473 contributions["overlap_fraction"][m] = cell_region.area / cell_box.area
474 contributions["weight"][m] = weight
475 # PSFs are expensive to evaluate; not clear if these columns
476 # are useful enough to be worth the cost.
477 contributions["psf_shape_flag"][m] = True
478 contributions["psf_shape_xx"][m] = np.nan
479 contributions["psf_shape_yy"][m] = np.nan
480 contributions["psf_shape_xy"][m] = np.nan
481 m += 1
482 n += 1
483 contributions = contributions[:m].copy()
484 inputs = inputs[:n].copy()
485 return CoaddProvenance(inputs, contributions)
488class CoaddProvenanceSerializationModel(ArchiveTree):
489 """A Pydantic model used to represent a serialized `CoaddProvenance`.
491 Notes
492 -----
493 We can't rewrite the Astropy tables directly into the archive (e.g. as
494 FITS binary tables for a FITS archive), because:
496 - `str` columns are a huge pain in both Numpy and FITS;
497 - the polygon columns need to be rewritten as array-valued columns.
499 To deal with the string columns (``instrument`` and ``physical_filter``)
500 we do dictionary compression: we map each distinct value of those columns
501 to an integer, and then we save that mapping to the model while saving
502 an integer version of that column in the table. But if there is actually
503 only one value in that column (the most common case by far) we just drop
504 the column and store that value directly in the model.
505 """
507 SCHEMA_NAME: ClassVar[str] = "coadd_provenance"
508 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
509 MIN_READ_VERSION: ClassVar[int] = 1
510 PUBLIC_TYPE: ClassVar[type] = CoaddProvenance
512 instrument: str | dict[str, int] = pydantic.Field(
513 description=(
514 "Instrument name for all inputs to this coadd, or a mapping from "
515 "instrument name to the integer used in its place in the tables."
516 )
517 )
518 physical_filter: str | dict[str, int] = pydantic.Field(
519 description="Physical filter name for all inputs to this coadd."
520 )
521 inputs: TableModel = pydantic.Field(description="Table of all inputs to the coadd.")
522 contributions: TableModel = pydantic.Field(description="Table of per-cell contributions to the coadd.")
524 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> CoaddProvenance:
525 """Deserialize a provenance from an input archive.
527 Parameters
528 ----------
529 archive
530 Archive to read from.
531 **kwargs
532 Unsupported keyword arguments are accepted only to provide
533 better error messages (raising
534 `.serialization.InvalidParameterError`).
536 Notes
537 -----
538 While `CoaddProvenance.subset` can be used to filter provenance
539 information down to just certain cells, there is no advantage to be
540 had from doing this during deserialization (the table data is not
541 ordered by cell, and hence there's read-slicing we can do).
542 """
543 if kwargs: 543 ↛ 544line 543 didn't jump to line 544 because the condition on line 543 was never true
544 raise InvalidParameterError(f"Unrecognized parameters for CoaddProvenance: {set(kwargs.keys())}.")
545 inputs = archive.get_table(self.inputs)
546 contributions = archive.get_table(self.contributions)
547 CoaddProvenanceSerializationModel._fix_str_for_deserialization(
548 "instrument", self.instrument, inputs, contributions
549 )
550 CoaddProvenanceSerializationModel._fix_str_for_deserialization(
551 "physical_filter", self.physical_filter, inputs
552 )
553 CoaddProvenanceSerializationModel._fix_polygon_for_deserialization(inputs)
554 for name, _, description in CoaddProvenance._INPUT_TABLE_COLUMNS:
555 inputs.columns[name].description = description
556 for name, _, description, unit in CoaddProvenance._CONTRIBUTION_TABLE_COLUMNS:
557 contributions.columns[name].description = description
558 contributions.columns[name].unit = unit
559 return CoaddProvenance(inputs=inputs, contributions=contributions)
561 @staticmethod
562 def _fix_str_for_serialization(column: str, *tables: astropy.table.Table) -> str | dict[str, int]:
563 """Rewrite a string column as an integer column or drop it.
565 Parameters
566 ----------
567 column
568 Name of the column to rewrite.
569 *tables
570 One or more astropy tables to rewrite. The first table is assumed
571 to have all values for this column that might appear in any other
572 tables.
574 Returns
575 -------
576 `str` | `dict` [`str`, `int`]
577 If there is only one unique value for this column in the first
578 table, that value (and the column will have been dropped from
579 all givne tables). If the tables are empty, the column is
580 dropped and an empty `dict` is returned. In all other cases the
581 given column is replaced with an integer column in all given
582 tables and the mapping from strings to integers is returned.
583 """
584 result: str | dict[str, int] = {name: n for n, name in enumerate(sorted(set(tables[0][column])))}
585 match len(result):
586 case 0: 586 ↛ 587line 586 didn't jump to line 587 because the pattern on line 586 never matched
587 pass
588 case 1: 588 ↛ 590line 588 didn't jump to line 590 because the pattern on line 588 always matched
589 (result,) = result.keys() # type: ignore[union-attr]
590 case _:
591 for table in tables:
592 table.columns[column] = astropy.table.Column(
593 data=[result[k] for k in table.columns[column]],
594 name=column,
595 dtype=np.uint8,
596 description=f"Integer mapped to {column} name.",
597 )
598 return result
599 # If we didn't remap to an integer (case 0 and 1 above), delete the
600 # column.
601 for table in tables:
602 del table.columns[column]
603 return result
605 @staticmethod
606 def _fix_str_for_deserialization(
607 column: str, value: str | dict[str, int], *tables: astropy.table.Table
608 ) -> None:
609 """Rewrite an integer column back to a string one.
611 Parameters
612 ----------
613 column
614 Name of the column to rewrite.
615 value
616 Value or mapping of values returned by
617 `_fix_str_for_serialization`.
618 tables
619 Tables to rewrite this column in.
620 """
621 match value:
622 case str(): 622 ↛ 625line 622 didn't jump to line 625 because the pattern on line 622 always matched
623 for table in tables:
624 table.columns[column] = astropy.table.Column([value] * len(table), dtype=object)
625 case dict():
626 mapping = {v: k for k, v in value.items()}
627 for table in tables:
628 table.columns[column] = astropy.table.Column(
629 [mapping[k] for k in table[column]], dtype=object
630 )
632 @staticmethod
633 def _fix_polygon_for_serialization(inputs: astropy.table.Table) -> None:
634 """Rewrite a polygon `object` column as a pair of array-valued columns
635 and an array-size column.
637 Parameters
638 ----------
639 inputs
640 A copy of the in-memory coadd inputs table to modify in-place into
641 its serialization form.
642 """
643 # Region intersections are sometimes represented as boxes rather
644 # than polygons, so normalize to polygons before writing vertex
645 # columns.
646 polygons = [p.to_polygon() if isinstance(p, Box) else p for p in inputs["polygon"]]
647 max_n_vertices = max((p.n_vertices for p in polygons), default=0)
648 inputs["n_vertices"] = astropy.table.Column(
649 [p.n_vertices for p in polygons],
650 name="n_vertices",
651 dtype=np.uint8,
652 description="Number of polygon vertices.",
653 )
654 inputs["x_vertices"] = astropy.table.Column(
655 name="x_vertices",
656 dtype=np.float64,
657 length=len(inputs),
658 shape=(max_n_vertices,),
659 description="X coordinates of polygon vertices, in tract coordinates.",
660 )
661 inputs["x_vertices"][:, :] = np.nan
662 inputs["y_vertices"] = astropy.table.Column(
663 name="y_vertices",
664 dtype=np.float64,
665 length=len(inputs),
666 shape=(max_n_vertices,),
667 description="Y coordinates of polygon vertices, in tract coordinates.",
668 )
669 inputs["y_vertices"][:, :] = np.nan
670 for i, polygon in enumerate(polygons):
671 inputs["n_vertices"][i] = polygon.n_vertices
672 inputs["x_vertices"][i][: polygon.n_vertices] = polygon.x_vertices
673 inputs["y_vertices"][i][: polygon.n_vertices] = polygon.y_vertices
674 del inputs["polygon"]
676 @staticmethod
677 def _fix_polygon_for_deserialization(inputs: astropy.table.Table) -> None:
678 """Rewrite a a pair of array-valued columns and an array-size column
679 into a polygon `object` column.
681 Parameters
682 ----------
683 inputs
684 The serialized version of the coadd inputs table, to be modified
685 in-place into its in-memory form.
686 """
687 polygons = [
688 Polygon(x_vertices=x_vertices[:n_vertices], y_vertices=y_vertices[:n_vertices])
689 for n_vertices, x_vertices, y_vertices in zip(
690 inputs["n_vertices"], inputs["x_vertices"], inputs["y_vertices"]
691 )
692 ]
693 del inputs["n_vertices"]
694 del inputs["x_vertices"]
695 del inputs["y_vertices"]
696 inputs["polygon"] = astropy.table.Column(polygons, name="polygon", dtype=np.object_)
699def _extract_day_obs_from_visit(visit_id: int) -> int:
700 """If the given visit ID looks like an ISO date-as-integer concatenated
701 with a sequence number, extract and return the date part.
703 Raises
704 ------
705 NotImplementedError
706 Raised if the visit ID does not look date-based.
707 """
708 id_str = str(visit_id)
709 if len(id_str) < 8:
710 raise NotImplementedError(f"visit ID {visit_id} does not start with a date.")
711 year = int(id_str[:4])
712 month = int(id_str[4:6])
713 day = int(id_str[6:8])
714 if year >= 1900 and year < 3000 and month <= 12 and day <= 31:
715 return (year * 100 + month) * 100 + day
716 raise NotImplementedError(f"visit ID {visit_id} does not start with a date.")