diff --git a/diagrams/erdiagram_targetdb_latest.pdf b/diagrams/erdiagram_targetdb_latest.pdf index 0091ba3..1a76a9e 100644 Binary files a/diagrams/erdiagram_targetdb_latest.pdf and b/diagrams/erdiagram_targetdb_latest.pdf differ diff --git a/docs/tbls/public.fluxstd.md b/docs/tbls/public.fluxstd.md index 3ca7a85..ba7ef39 100644 --- a/docs/tbls/public.fluxstd.md +++ b/docs/tbls/public.fluxstd.md @@ -102,8 +102,8 @@ | fluxstd_pkey | CREATE UNIQUE INDEX fluxstd_pkey ON public.fluxstd USING btree (fluxstd_id) | | uq_obj_id_input_catalog_id_version | CREATE UNIQUE INDEX uq_obj_id_input_catalog_id_version ON public.fluxstd USING btree (obj_id, input_catalog_id, version) | | fluxstd_q3c_ang2ipix_idx | CREATE INDEX fluxstd_q3c_ang2ipix_idx ON public.fluxstd USING btree (q3c_ang2ipix(ra, "dec")) | -| ix_fluxstd_version | CREATE INDEX ix_fluxstd_version ON public.fluxstd USING btree (version) | | ix_fluxstd_input_catalog_fluxstdid | CREATE INDEX ix_fluxstd_input_catalog_fluxstdid ON public.fluxstd USING btree (input_catalog_id, fluxstd_id) | +| ix_fluxstd_version | CREATE INDEX ix_fluxstd_version ON public.fluxstd USING btree (version) | | ix_fluxstd_version_fluxstdid | CREATE INDEX ix_fluxstd_version_fluxstdid ON public.fluxstd USING btree (version, fluxstd_id) | ## Relations diff --git a/docs/tbls/public.sky.md b/docs/tbls/public.sky.md index f4d93bf..6b4f98d 100644 --- a/docs/tbls/public.sky.md +++ b/docs/tbls/public.sky.md @@ -34,9 +34,9 @@ | ---- | ---------- | | sky_pkey | CREATE UNIQUE INDEX sky_pkey ON public.sky USING btree (sky_id) | | sky_obj_id_input_catalog_id_version_key | CREATE UNIQUE INDEX sky_obj_id_input_catalog_id_version_key ON public.sky USING btree (obj_id, input_catalog_id, version) | +| ix_sky_input_catalog_id | CREATE INDEX ix_sky_input_catalog_id ON public.sky USING btree (input_catalog_id) | | sky_q3c_ang2ipix_idx | CREATE INDEX sky_q3c_ang2ipix_idx ON public.sky USING btree (q3c_ang2ipix(ra, "dec")) | | ix_sky_version | CREATE INDEX ix_sky_version ON public.sky USING btree (version) | -| ix_sky_input_catalog_id | CREATE INDEX ix_sky_input_catalog_id ON public.sky USING btree (input_catalog_id) | ## Relations diff --git a/docs/tbls/public.target.md b/docs/tbls/public.target.md index 1728dae..12ceb5a 100644 --- a/docs/tbls/public.target.md +++ b/docs/tbls/public.target.md @@ -121,11 +121,11 @@ | target_pkey | CREATE UNIQUE INDEX target_pkey ON public.target USING btree (target_id) | | target_propid_obcode_catid_objid_resolution_key | CREATE UNIQUE INDEX target_propid_obcode_catid_objid_resolution_key ON public.target USING btree (proposal_id, ob_code, input_catalog_id, obj_id, is_medium_resolution) | | target_propid_obcode_key | CREATE UNIQUE INDEX target_propid_obcode_key ON public.target USING btree (proposal_id, ob_code) | +| target_proposal_id_idx | CREATE INDEX target_proposal_id_idx ON public.target USING btree (proposal_id) | +| target_obj_id_input_catalog_id_idx | CREATE INDEX target_obj_id_input_catalog_id_idx ON public.target USING btree (obj_id, input_catalog_id) | | target_input_catalog_id_idx | CREATE INDEX target_input_catalog_id_idx ON public.target USING btree (input_catalog_id) | | target_proposal_id_obj_id_idx | CREATE INDEX target_proposal_id_obj_id_idx ON public.target USING btree (proposal_id, obj_id) | | target_q3c_ang2ipix_idx | CREATE INDEX target_q3c_ang2ipix_idx ON public.target USING btree (q3c_ang2ipix(ra, "dec")) | -| target_proposal_id_idx | CREATE INDEX target_proposal_id_idx ON public.target USING btree (proposal_id) | -| target_obj_id_input_catalog_id_idx | CREATE INDEX target_obj_id_input_catalog_id_idx ON public.target USING btree (obj_id, input_catalog_id) | ## Relations diff --git a/src/targetdb/utils.py b/src/targetdb/utils.py index 78c8d97..9b75dce 100644 --- a/src/targetdb/utils.py +++ b/src/targetdb/utils.py @@ -371,6 +371,101 @@ def draw_diagram( raise ValueError(f"Unsupported generator: {generator}") +FILTER_BANDS = ["u", "v", "g", "r", "i", "z", "y", "j"] + + +def normalize_filter_columns(df): + """ + Normalize filter_* columns (e.g., filter_g, filter_i) in place so that + missing values are stored as None instead of being passed through to the + database as-is. + + filter_* columns are ForeignKey columns referencing filter_name.filter_name. + Depending on how the input was loaded, a missing value may show up as a + float NaN (e.g., from a masked ecsv column) or as an empty string (e.g., + from a CSV column read with keep_default_na=False). Either one fails the + foreign key constraint if inserted as-is (NaN is even coerced to the + literal string "NaN" by the database driver), so both are normalized to + None here. + + Parameters + ---------- + df : pandas.DataFrame + The DataFrame to normalize. Modified in place. + + Returns + ------- + df : pandas.DataFrame + The same DataFrame, for convenience. + """ + for band in FILTER_BANDS: + col = f"filter_{band}" + if col not in df.columns: + continue + df.loc[df[col].isna() | (df[col] == ""), col] = None + return df + + +FLUX_COLUMN_PREFIXES = ["", "psf_", "total_"] + + +def check_filter_flux_consistency(df): + """ + Raise an error if a band has a flux value but no filter. + + A flux (or flux error) measurement is meaningless without knowing which + filter it was measured through, so filter_{band} being None while + flux_{band} (or psf_flux_{band}, total_flux_{band}, and their + *_error_* counterparts) is set indicates a problem upstream in the input + data rather than something to silently accept. + + Parameters + ---------- + df : pandas.DataFrame + The DataFrame to check. filter_* columns are expected to already be + normalized (see `normalize_filter_columns`), i.e., missing filters + are represented as None rather than NaN or "". + + Returns + ------- + df : pandas.DataFrame + The same DataFrame, unmodified, for convenience. + + Raises + ------ + ValueError + If any row has a flux value for a band whose filter is None. + """ + for band in FILTER_BANDS: + filter_col = f"filter_{band}" + if filter_col not in df.columns: + continue + + missing_filter = df[filter_col].isna() + if not missing_filter.any(): + continue + + for prefix in FLUX_COLUMN_PREFIXES: + for kind in ["flux", "flux_error"]: + flux_col = f"{prefix}{kind}_{band}" + if flux_col not in df.columns: + continue + + inconsistent = missing_filter & df[flux_col].notna() + if inconsistent.any(): + bad_indices = df.index[inconsistent].tolist() + logger.error( + f"{flux_col} has values but {filter_col} is missing " + f"for rows {bad_indices}." + ) + raise ValueError( + f"{flux_col} has values but {filter_col} is missing " + f"for rows {bad_indices}. A flux value without a " + "filter is not allowed." + ) + return df + + def join_backref_values(df, db=None, table=None, key=None, check_key=None): """ Joins a DataFrame with a table from a database on a specified key and checks for non-existing keys. @@ -445,6 +540,8 @@ def add_backref_values(df, db=None, table=None, upload_id=None): """ df_tmp = df.copy() + normalize_filter_columns(df_tmp) + check_filter_flux_consistency(df_tmp) backref_tables, backref_keys, backref_check_keys = [], [], [] if table == "target": @@ -649,15 +746,10 @@ def make_target_df_from_uploader( logger.error(f"flux_type must be 'total' or 'psf'. {flux_type=}") raise ValueError(f"flux_type must be 'total' or 'psf'. {flux_type=}") - # fill missing values with None or NaN for filters and fluxes + normalize_filter_columns(df) + + # rename flux columns to match the flux_type for band in ["g", "r", "i", "z", "y", "j"]: - if f"filter_{band}" in df.columns: - # if the table is a masked table, fill the masked values with None for filters - # if the table is not a masked table, just pass - try: - df.loc[df[f"filter_{band}"].isna(), f"filter_{band}"] = None - except AttributeError: - pass if f"flux_{band}" in df.columns: logger.info(f"flux_{band} is renamed to {flux_type}_flux_{band}") df.rename( diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..f22856e --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python + +import numpy as np +import pandas as pd +import pytest + +from targetdb.utils import add_backref_values, check_filter_flux_consistency + + +def test_add_backref_values_normalizes_missing_filter_values_for_fluxstd(): + # filter_g/filter_i are ForeignKey columns to filter_name.filter_name. + # Empty entries in the input (NaN from a masked ecsv column, or an empty + # string from a CSV column read with keep_default_na=False) must be + # normalized to None, otherwise the literal "NaN"/"" is sent to the + # database and violates the foreign key constraint. This must hold for + # any table that goes through add_backref_values (fluxstd, sky, + # user_pointing, and target when not coming from the uploader), not just + # for the uploader-specific target code path. + df = pd.DataFrame( + { + "input_catalog_id": [1, 1, 1], + "filter_g": ["g_sdss", np.nan, ""], + "filter_i": ["i_sdss", "i_sdss", "i_sdss"], + } + ) + + df_out = add_backref_values(df, db=None, table="fluxstd") + + assert df_out["filter_g"].tolist() == ["g_sdss", None, None] + assert df_out["filter_i"].tolist() == ["i_sdss", "i_sdss", "i_sdss"] + + # None must round-trip as None (not float NaN) through to_dict, since + # that is what actually gets handed to bulk_insert_mappings. + records = df_out.to_dict(orient="records") + assert records[1]["filter_g"] is None + assert records[2]["filter_g"] is None + + +def test_add_backref_values_raises_when_flux_present_without_filter(): + # A flux value is meaningless without knowing which filter it was + # measured through, so filter_g missing while psf_flux_g is set must be + # rejected rather than silently inserted. + df = pd.DataFrame( + { + "input_catalog_id": [1, 1], + "filter_g": ["g_sdss", np.nan], + "psf_flux_g": [123.4, 56.7], + } + ) + + with pytest.raises(ValueError, match="psf_flux_g"): + add_backref_values(df, db=None, table="fluxstd") + + +@pytest.mark.parametrize( + "flux_col", ["flux_g", "psf_flux_g", "total_flux_g", "psf_flux_error_g"] +) +def test_check_filter_flux_consistency_raises_for_each_flux_column_kind(flux_col): + df = pd.DataFrame({"filter_g": [None], flux_col: [1.0]}) + + with pytest.raises(ValueError, match=flux_col): + check_filter_flux_consistency(df) + + +def test_check_filter_flux_consistency_allows_missing_filter_and_flux(): + df = pd.DataFrame({"filter_g": [None, "g_sdss"], "psf_flux_g": [np.nan, 1.0]}) + + # should not raise + check_filter_flux_consistency(df)