diff --git a/config.py b/config.py index b508efb..4388e2c 100644 --- a/config.py +++ b/config.py @@ -101,11 +101,31 @@ class Columns(BaseModel): class Param(BaseModel): combine_type: str = "concatenate" extra_columns: dict[str, Any] = Field(default_factory=dict) - # Zero disables the cut; valid active cuts are 1, 2, 3, 4, 5, 6. + # Zero disables the cut; valid active cuts are 1, 2, 3, 4. z_flag_homogenized_value_to_cut: float = 3.0 + include_unclassified: bool = True + include_galaxy: bool = True + include_star: bool = False + include_agn: bool = True + include_qso: bool = True + include_galactic: bool = False flags_translation_file: str = str(Path(MAINDIR, "flags_translation.yaml")) insert_DP1_footprint_flag: bool = False - insert_rubin_footprint_flag: bool = True + insert_rubin_footprint_flag: bool = False + + @model_validator(mode="after") + def validate_object_type_inclusion(self): + inclusion = ( + self.include_unclassified, + self.include_galaxy, + self.include_star, + self.include_agn, + self.include_qso, + self.include_galactic, + ) + if not any(inclusion): + raise ValueError("at least one include_* object-type option must be true") + return self class Config(BaseModel): diff --git a/config.template.yaml b/config.template.yaml index dd92a79..276cff4 100644 --- a/config.template.yaml +++ b/config.template.yaml @@ -87,7 +87,15 @@ param: # ORIGINAL_ID: # source: id # type: str - z_flag_homogenized_value_to_cut: 0.0 # 0 disables the cut; valid active cuts are 1, 2, 3, 4, 5, 6. + z_flag_homogenized_value_to_cut: 0.0 # 0 disables the cut; valid active cuts are 1, 2, 3, 4. + # Object types allowed to participate in the deduplication graph. Disabled + # types remain in marked outputs as isolated rows with tie_result=3. + include_unclassified: true + include_galaxy: true + include_star: false + include_agn: true + include_qso: true + include_galactic: false insert_DP1_footprint_flag: false # Adds is_in_DP1_fields (1/0) - insert_rubin_footprint_flag: true # Adds is_in_rubin_footprint (1/0) + insert_rubin_footprint_flag: false # Adds is_in_rubin_footprint (1/0) flags_translation_file: flags_translation.yaml # File with homogenization rules for z_flag and type diff --git a/flags_translation.yaml b/flags_translation.yaml index ba8ab7c..6c4802e 100644 --- a/flags_translation.yaml +++ b/flags_translation.yaml @@ -13,16 +13,22 @@ tiebreaking_priority: # contain at least one valid numeric value in every input catalog. # The columns will be evaluated in the order listed to resolve duplicates. # In deduplication modes, z_flag_homogenized is still created and preserved - # for star classification (value 6), even when omitted from this ranking. + # as a quality field even when omitted from this ranking. delta_z_threshold: 0.001 # Maximum allowed redshift difference (|z1 - z2|) to resolve hard ties # Applied only when initial tiebreaking via columns results in a tie (tie_result == 2) crossmatch_radius_arcsec: 0.5 # Angular radius (in arcseconds) used to perform spatial crossmatching. +max_representative_radius_arcsec: 1.0 # Split transitive components so every member stays within this radius of its ranked reference; null disables. margin_threshold_arcsec: 5.0 # HATS margin used by independent partition-local deduplication. margin_warning_fraction: 0.8 # Warn when a boundary component spans this fraction of the margin. validate_global_graph_edges: false # Distributed edge/group consistency check (costs two joins). validate_global_tie_invariants: true # Fail on invalid winner/hard-tie patterns after label merge. +tie_invariant_diagnostics_enabled: true # Cheap aggregate reason counts for invalid tie groups. +tie_invariant_diagnostics_detailed_enabled: false # Expensive sample of group IDs and member CRD_IDs. +tie_invariant_diagnostics_sample_size: 10 # Maximum invalid groups in the detailed sample. +tie_invariant_diagnostics_max_rows: 100 # Maximum member rows collected across sampled groups. +label_merge_diagnostics_enabled: true # Cheap count of rows without a newly computed dedup label. validate_crd_id_uniqueness: true # Global nunique validation for every prepared input catalog. repartition_prepared_catalogs: false # Extra full-data pass to resize prepared partitions. prepared_partition_size: "256MB" # Used only when repartition_prepared_catalogs is enabled. @@ -33,14 +39,14 @@ crossmatch_saturation_warn_fraction: 0.01 # Warn when >=1% of all source object crossmatch_saturation_fail_fraction: 0.10 # Fail at >=10%; set null to disable failure. crossmatch_geometry_diagnostics_enabled: false # Separation/component-shape diagnostics over materialized matches. representative_radius_diagnostics_enabled: false # Component radius/extent diagnostics during and after deduplication. -dedup_edge_diagnostics_enabled: false # Additional star-edge diagnostics while building the deduplication graph. +dedup_edge_diagnostics_enabled: false # Additional diagnostics for object types excluded while building the deduplication graph. instrument_type_priority: s: 3 # Spectroscopic (most reliable type) g: 2 # Grism p: 1 # Photometric (least reliable) -# === Survey-specific translation rules for z_flag and type === +# === Survey-specific translation rules for z_flag, instrument, and object type === # These are our internal definitions used to harmonize redshift quality and instrument types across surveys. # The quality flag system is inspired by the VVDS survey but adapted to our needs: # @@ -50,12 +56,27 @@ instrument_type_priority: # - 2: 70% < Confidence < 90% # - 3: 90% < Confidence < 99% # - 4: Confidence > 99% -# - 6: Object identified as non-extragalactic (e.g., stars) +# - null: quality unavailable or not applicable; stellar/galactic rows without +# an independently documented quality remain null # Each block corresponds to a survey name and includes: -# - z_flag_translation: how to map original quality flags or based on conditions into a standardized quality scale (0-6) +# - z_flag_translation: how to map original quality flags or based on conditions into a standardized quality scale (0-4) # - instrument_type_translation: how to map original types into a standard label (s, g, p) +# - object_type_translation: how to map classifications into star, galactic, +# qso, agn, or galaxy. `galactic` means a confirmed Galactic/non-extragalactic +# source without a secure stellar subtype. `qso` requires an explicit +# QSO/quasar classification; `agn` is used for generic AGN and BLAGN +# classifications without an explicit QSO label. +# Its default is null; catalogs without object classifications are valid. +# A rule may declare `source: ` and `optional_source: true` to +# discover a non-standard input column without a frontend columns mapping. # You can mix value-based mappings and conditional rules (using Python expressions) +# Conditions are evaluated in YAML order. When multiple conditions match a row, +# the later condition takes precedence ("last match wins"). Conflicting matches +# emit a warning unless that translation declares `allow_condition_overlap: true`. +# z_flag and instrument translations accept `fast_path: auto|disabled|required`. +# `auto` is the default, `disabled` forces YAML rules, and `required` fails when +# the input is not compatible with the corresponding probability/s-g-p fast path. save_expr_columns: false @@ -84,16 +105,31 @@ translation_rules: default: "s" 2DFLENS: + object_type_translation: + conditions: + - expr: "z_flag == 6" + value: "galactic" + default: null z_flag_translation: 1: 0 2: 1 3: 3 4: 4 - 6: 6 instrument_type_translation: default: "s" 2MRS: + object_type_translation: + conditions: + # ZCAT type -9 combines QSO and generic AGN; both belong to the broader + # AGN class, but the information is insufficient for the QSO subtype. + - expr: "str(TYPE)[:2] == '-9'" + value: "agn" + # Use only documented ZCAT galaxy morphology codes. Type 98 means that + # the source was never visually examined and is intentionally omitted. + - expr: "TYPE.str.contains(r'^(?:-[1-7]|20|19|16|15|11|10|[0-9])(?:[^0-9]|$)', regex=True, na=False)" + value: "galaxy" + default: null z_flag_translation: conditions: - expr: "z_err == 0" @@ -107,10 +143,15 @@ translation_rules: default: "s" 3D-HST: + object_type_translation: + conditions: + - expr: "z_best_s == 0" + value: "star" + default: null z_flag_translation: conditions: - expr: "z_best_s == 0" - value: 6 + value: null - expr: "z_best_s == 1 and z_spec != -1" value: 3 - expr: "z_best_s == 2 and use_zgrism == 1 and flag1 == 0 and flag2 == 0" @@ -129,12 +170,16 @@ translation_rules: default: "g" 6DFGS: + object_type_translation: + conditions: + - expr: "z_flag == 6" + value: "galactic" + default: null z_flag_translation: 1: 0 2: 1 3: 3 4: 4 - 6: 6 instrument_type_translation: default: "s" @@ -179,12 +224,20 @@ translation_rules: default: "s" COSMOS2020_CLASSIC: + object_type_translation: + conditions: + - expr: "lp_type == 0" + value: "galaxy" + - expr: "lp_type == 1" + value: "star" + default: null z_flag_translation: + allow_condition_overlap: true conditions: - expr: "lp_type == -9 or not np.isfinite(z)" value: 0 - expr: "lp_type == 1" - value: 6 + value: null - expr: "(lp_type != 1 and lp_type != -9) and np.isfinite(z)" value: 1 - expr: "(lp_type != 1 and lp_type != -9) and np.isfinite(z) and np.isfinite(lp_zPDF) and np.isfinite(lp_zMinChi2) and np.isfinite(lp_zPDF_l68) and np.isfinite(lp_zPDF_u68) and lp_zPDF_l68 <= z <= lp_zPDF_u68 and lp_zPDF_l68 <= lp_zPDF <= lp_zPDF_u68 and lp_zPDF_l68 <= lp_zMinChi2 <= lp_zPDF_u68" @@ -196,6 +249,11 @@ translation_rules: default: "p" COSMOS_SRC: + object_type_translation: + conditions: + - expr: "10 <= z_flag < 20" + value: "agn" + default: null z_flag_translation: conditions: - expr: "np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 0" @@ -215,23 +273,44 @@ translation_rules: default: "s" COSMOS_Web: + object_type_translation: + conditions: + - expr: "type == '0'" + value: "galaxy" + - expr: "type == '1'" + value: "star" + - expr: "type == '2'" + value: "qso" + default: null z_flag_translation: conditions: - - expr: "type != 1 and z_flag != 0" + - expr: "type != '1' and z_flag != 0" value: 0 - - expr: "type != 1 and z_flag == 0 and np.abs(zpdf_med - zchi2) > 0.001" + - expr: "type != '1' and z_flag == 0 and np.abs(zpdf_med - zchi2) > 0.001" value: 1 - - expr: "type != 1 and z_flag == 0 and np.abs(zpdf_med - zchi2) <= 0.001 and nbfilt < 30" + - expr: "type != '1' and z_flag == 0 and np.abs(zpdf_med - zchi2) <= 0.001 and nbfilt < 30" value: 2 - - expr: "type != 1 and z_flag == 0 and np.abs(zpdf_med - zchi2) <= 0.001 and nbfilt > 30" + - expr: "type != '1' and z_flag == 0 and np.abs(zpdf_med - zchi2) <= 0.001 and nbfilt >= 30" value: 3 - - expr: "type == 1" - value: 6 + - expr: "type == '1'" + value: null default: 0 instrument_type_translation: default: "p" DEIMOS_10K: + object_type_translation: + allow_condition_overlap: true + conditions: + - expr: "10 <= z_flag < 20" + value: "agn" + - expr: "Remarks.str.strip().str.lower().isin(['star', 'm star'])" + value: "star" + - expr: "Remarks.str.contains(',star$', case=False, na=False, regex=True)" + value: "star" + - expr: "Remarks.str.contains('[(]QSO[)]|,QSO$', case=False, na=False, regex=True)" + value: "qso" + default: null z_flag_translation: conditions: - expr: "np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 0" @@ -251,6 +330,15 @@ translation_rules: default: "s" DESI: + object_type_translation: + conditions: + - expr: "SPECTYPE == 'STAR'" + value: "star" + - expr: "SPECTYPE == 'QSO'" + value: "qso" + - expr: "SPECTYPE == 'GALAXY'" + value: "galaxy" + default: null z_flag_translation: conditions: - expr: "SPECTYPE != 'STAR' and z_flag != 0 and ZCAT_PRIMARY != True" @@ -262,12 +350,21 @@ translation_rules: - expr: "SPECTYPE != 'STAR' and z_flag == 0 and ZCAT_PRIMARY == True and z_err < 0.001" value: 4 - expr: "SPECTYPE == 'STAR'" - value: 6 + value: null default: 0 instrument_type_translation: default: "s" DESI_DR1: + object_type_translation: + conditions: + - expr: "SPECTYPE == 'STAR'" + value: "star" + - expr: "SPECTYPE == 'QSO'" + value: "qso" + - expr: "SPECTYPE == 'GALAXY'" + value: "galaxy" + default: null z_flag_translation: conditions: - expr: "SPECTYPE != 'STAR' and z_flag != 0 and ZCAT_PRIMARY != True" @@ -279,12 +376,21 @@ translation_rules: - expr: "SPECTYPE != 'STAR' and z_flag == 0 and ZCAT_PRIMARY == True and z_err < 0.001" value: 4 - expr: "SPECTYPE == 'STAR'" - value: 6 + value: null default: 0 instrument_type_translation: default: "s" DESI_DEEP_PILOT: + object_type_translation: + conditions: + - expr: "SPECTYPE == 'STAR'" + value: "star" + - expr: "SPECTYPE == 'QSO'" + value: "qso" + - expr: "SPECTYPE == 'GALAXY'" + value: "galaxy" + default: null z_flag_translation: conditions: - expr: "np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 0" @@ -302,13 +408,23 @@ translation_rules: default: "s" ELAISS1OID: + object_type_translation: + conditions: + - expr: "Class == 1" + value: "agn" + - expr: "Class == 5" + value: "agn" + - expr: "Class == 2 or Class == 3 or Class == 6" + value: "galaxy" + - expr: "Class == 4" + value: "star" + default: null z_flag_translation: 0: 0 1: 1 2: 2 3: 3 4: 4 - 6: 6 instrument_type_translation: conditions: - expr: "zfinal_type == 's'" @@ -318,13 +434,21 @@ translation_rules: default: "p" ELAISFBMC: + object_type_translation: + conditions: + - expr: "tSp == 1 or tSp == 2 or tSp == 3" + value: "galaxy" + - expr: "tSp == 4 or tSp == 5 or tSp == 6 or tSp == 8" + value: "agn" + - expr: "tSp == 7" + value: "star" + default: null z_flag_translation: 0: 0 1: 1 2: 2 3: 3 4: 4 - 6: 6 instrument_type_translation: conditions: - expr: "zbest_type == 's'" @@ -334,12 +458,25 @@ translation_rules: default: "p" ELG_FIGS: + object_type_translation: + default: "galaxy" z_flag_translation: default: 3 instrument_type_translation: default: "s" EUCLID_Q1: + # spe_class is discovered directly in the raw dataframe by the backend. + # The current production catalog does not contain it, so its absence must + # safely produce null output without requiring a frontend column mapping. + object_type_translation: + source: spe_class + optional_source: true + STAR: "star" + GALAXY: "galaxy" + QSO: "qso" + UNDEF: null + default: null z_flag_translation: conditions: - expr: "z_flag == 0.0" @@ -357,6 +494,11 @@ translation_rules: default: "s" FMOS_COSMOS: + object_type_translation: + conditions: + - expr: "z_flag == -1" + value: "star" + default: null z_flag_translation: -99: 0 0: 0 @@ -364,7 +506,6 @@ translation_rules: 2: 2 3: 3 4: 4 - -1: 6 instrument_type_translation: default: "s" @@ -379,6 +520,12 @@ translation_rules: default: "s" HELP-DMU23: + object_type_translation: + conditions: + # The HELP flag combines QSO and AGN, so only the broader class is safe. + - expr: "agn_flag == 1" + value: "agn" + default: null z_flag_translation: 1: 0 2: 1 @@ -400,6 +547,7 @@ translation_rules: JADES_DR5_PHOTOZ: z_flag_translation: + allow_condition_overlap: true conditions: - expr: "not np.isfinite(z) or not np.isfinite(z_ml) or not np.isfinite(chi_a) or not np.isfinite(nfilt) or nfilt <= 0" value: 0 @@ -415,6 +563,7 @@ translation_rules: JADES_DR5_PHOTOZ_KRON: z_flag_translation: + allow_condition_overlap: true conditions: - expr: "not np.isfinite(z) or not np.isfinite(z_ml) or not np.isfinite(chi_a) or not np.isfinite(nfilt) or nfilt <= 0" value: 0 @@ -442,16 +591,31 @@ translation_rules: default: "s" OZDES: + object_type_translation: + conditions: + # Object_types contains targeting categories, not confirmed object + # classifications. qop=6 is explicitly a securely classified star. + - expr: "z_flag == 6" + value: "star" + default: null z_flag_translation: 1: 0 2: 1 3: 3 4: 4 - 6: 6 instrument_type_translation: default: "s" PRIMUS: + object_type_translation: + conditions: + - expr: "CLASS == 'GALAXY'" + value: "galaxy" + - expr: "CLASS == 'STAR'" + value: "star" + - expr: "CLASS == 'AGN'" + value: "agn" + default: null z_flag_translation: -1: 0 2: 1 @@ -461,6 +625,15 @@ translation_rules: default: "g" SDSS_DR17: + object_type_translation: + conditions: + - expr: "CLASS == 'STAR'" + value: "star" + - expr: "CLASS == 'QSO'" + value: "qso" + - expr: "CLASS == 'GALAXY'" + value: "galaxy" + default: null z_flag_translation: conditions: - expr: "CLASS != 'STAR' and z_flag != 0 and SPECPRIMARY != 1" @@ -472,12 +645,21 @@ translation_rules: - expr: "CLASS != 'STAR' and z_flag == 0 and SPECPRIMARY == 1 and z_err < 0.001" value: 4 - expr: "CLASS == 'STAR'" - value: 6 + value: null default: 0 instrument_type_translation: default: "s" SDSS_DR19: + object_type_translation: + conditions: + - expr: "CLASS == 'STAR'" + value: "star" + - expr: "CLASS == 'QSO'" + value: "qso" + - expr: "CLASS == 'GALAXY'" + value: "galaxy" + default: null z_flag_translation: conditions: - expr: "CLASS != 'STAR' and z_flag != 0 and SPECPRIMARY != 1" @@ -489,19 +671,29 @@ translation_rules: - expr: "CLASS != 'STAR' and z_flag == 0 and SPECPRIMARY == 1 and z_err < 0.001" value: 4 - expr: "CLASS == 'STAR'" - value: 6 + value: null default: 0 instrument_type_translation: default: "s" SWIRE-REVISED: + object_type_translation: + allow_condition_overlap: true + conditions: + - expr: "mst == -1" + value: "star" + - expr: "mst == 1" + value: "galaxy" + # Explicit QSO templates take precedence over the morphology scale. + - expr: "(13 <= J1 <= 30) or (13 <= J2 <= 15)" + value: "qso" + default: null z_flag_translation: 0: 0 1: 1 2: 2 3: 3 4: 4 - 6: 6 instrument_type_translation: conditions: - expr: "zfinal_type == 's'" @@ -511,6 +703,11 @@ translation_rules: default: "p" VANDELS: + object_type_translation: + conditions: + - expr: "np.trunc(float(z_flag)) == 13 or np.trunc(float(z_flag)) == 14 or np.trunc(float(z_flag)) == 19 or np.trunc(float(z_flag)) == 213 or np.trunc(float(z_flag)) == 214 or np.trunc(float(z_flag)) == 219" + value: "agn" + default: null z_flag_translation: conditions: - expr: "np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 0" @@ -530,6 +727,13 @@ translation_rules: default: "s" VIMOS: + object_type_translation: + conditions: + - expr: "COMM.str.strip().str.lower() == 'star'" + value: "star" + - expr: "COMM.str.contains('[(]BLAGN[)]', case=False, na=False, regex=True)" + value: "agn" + default: null z_flag_translation: 4: 4 3: 3 @@ -540,10 +744,15 @@ translation_rules: default: "s" VIPERS_PDR2: + object_type_translation: + conditions: + # classFlag=-1 is only a photometric "stellar-like" classification and + # is intentionally left null. Only spectroscopic BLAGN is retained. + - expr: "np.trunc(float(z_flag)) == 13 or np.trunc(float(z_flag)) == 14 or np.trunc(float(z_flag)) == 19 or np.trunc(float(z_flag)) == 213 or np.trunc(float(z_flag)) == 214 or np.trunc(float(z_flag)) == 219" + value: "agn" + default: null z_flag_translation: conditions: - - expr: "classFlag == -1" - value: 6 - expr: "classFlag != -1 and np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 1" value: 1 - expr: "classFlag != -1 and np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 2" @@ -559,6 +768,11 @@ translation_rules: default: "s" VUDS: + object_type_translation: + conditions: + - expr: "10 <= z_flag < 20" + value: "agn" + default: null z_flag_translation: conditions: - expr: "np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 1" @@ -576,6 +790,11 @@ translation_rules: default: "s" VVDS: + object_type_translation: + conditions: + - expr: "10 <= z_flag < 20" + value: "agn" + default: null z_flag_translation: conditions: - expr: "np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 0" @@ -595,10 +814,19 @@ translation_rules: default: "s" XMM_LSS: + object_type_translation: + conditions: + - expr: "Class == 'STA'" + value: "star" + - expr: "Class == 'QSO'" + value: "qso" + - expr: "Class == 'ALG' or Class == 'ELG'" + value: "galaxy" + default: null z_flag_translation: conditions: - expr: "Class == 'STA'" - value: 6 + value: null - expr: "Class != 'STA' and z_flag == 3" value: 1 - expr: "Class != 'STA' and z_flag == 2" @@ -610,6 +838,11 @@ translation_rules: default: "s" ZCOSMOS: + object_type_translation: + conditions: + - expr: "10 <= z_flag < 20" + value: "agn" + default: null z_flag_translation: conditions: - expr: "np.trunc(float(z_flag)) - 10*np.trunc(np.trunc(float(z_flag))/10) == 0" diff --git a/packages/crossmatch_cross.py b/packages/crossmatch_cross.py index 78018db..c1dcc8b 100644 --- a/packages/crossmatch_cross.py +++ b/packages/crossmatch_cross.py @@ -511,9 +511,11 @@ def _get_expr_schema_hints(translation_config: dict | None) -> dict: dict: Expr column schema hints or empty dict. """ cfg = translation_config or {} - if cfg.get("save_expr_columns") is False: - return {} - return cfg.get("expr_column_schema", {}) or {} + hints = {} + if cfg.get("save_expr_columns") is not False: + hints.update(cfg.get("expr_column_schema", {}) or {}) + hints.update(cfg.get("runtime_schema_hints", {}) or {}) + return hints def _ensure_compared_to(cat): @@ -546,6 +548,7 @@ def _ensure_compared_to(cat): "tie_result": DTYPE_INT8, "z_flag_homogenized": DTYPE_FLOAT, "instrument_type_homogenized": DTYPE_STR, + "object_type_homogenized": DTYPE_STR, "compared_to": DTYPE_STR, } diff --git a/packages/deduplication.py b/packages/deduplication.py index 77cb952..b470e1e 100644 --- a/packages/deduplication.py +++ b/packages/deduplication.py @@ -30,6 +30,7 @@ import pandas as pd REPRESENTATIVE_RADIUS_DIAGNOSTIC_COLUMN = "_diag_representative_max_radius_arcsec" +SPLIT_REFERENCE_COLUMN = "__split_reference_crd" # ----------------------- # Project @@ -76,10 +77,71 @@ def _phase_logger() -> logging.LoggerAdapter: "run_dedup_with_lsdb_map_partitions", "count_global_edge_group_mismatches", "count_global_tie_invariant_violations", + "build_global_tie_invariant_diagnostics", "filter_dask_by_tie_treatment", "filter_pandas_by_tie_treatment", + "validate_object_type_inclusion", ] +OBJECT_TYPE_INCLUDE_DEFAULTS = { + "include_unclassified": True, + "include_galaxy": True, + "include_star": False, + "include_agn": True, + "include_qso": True, + "include_galactic": False, +} +_OBJECT_TYPE_TO_INCLUDE_KEY = { + "galaxy": "include_galaxy", + "star": "include_star", + "agn": "include_agn", + "qso": "include_qso", + "galactic": "include_galactic", +} + + +def validate_object_type_inclusion(config: Mapping[str, object] | None) -> dict[str, bool]: + """Return strict, complete object-type inclusion settings.""" + supplied = dict(config or {}) + unknown = sorted(set(supplied) - set(OBJECT_TYPE_INCLUDE_DEFAULTS)) + if unknown: + raise ValueError(f"Unknown object-type inclusion option(s): {unknown}") + result = dict(OBJECT_TYPE_INCLUDE_DEFAULTS) + for key, value in supplied.items(): + if not isinstance(value, bool): + raise TypeError(f"param.{key} must be a boolean, got {type(value).__name__}") + result[key] = value + if not any(result.values()): + raise ValueError( + "At least one object-type inclusion option must be true: " + + ", ".join(OBJECT_TYPE_INCLUDE_DEFAULTS) + ) + return result + + +def _excluded_object_type_mask( + object_types: pd.Series, + inclusion: Mapping[str, bool], +) -> pd.Series: + normalized = object_types.astype("string").str.strip().str.lower() + included = pd.Series(False, index=object_types.index, dtype=bool) + included |= normalized.isna() & bool(inclusion["include_unclassified"]) + for object_type, key in _OBJECT_TYPE_TO_INCLUDE_KEY.items(): + included |= normalized.eq(object_type).fillna(False) & bool(inclusion[key]) + return ~included + + +def _effective_z_flag_score(df: pd.DataFrame) -> pd.Series: + """Quality ranking score; stellar classes without quality receive 0.5.""" + score = _to_numeric( + df.get("z_flag_homogenized", pd.Series(np.nan, index=df.index)) + ).astype("float64") + object_type = df.get( + "object_type_homogenized", pd.Series(pd.NA, index=df.index, dtype="string") + ).astype("string").str.strip().str.lower() + stellar_without_quality = object_type.isin(["star", "galactic"]) & score.isna() + return score.mask(stellar_without_quality, 0.5) + # ----------------------- # Small helpers: string/parse/score @@ -109,6 +171,7 @@ def validate_spatial_safety( crossmatch_radius_arcsec: float, margin_threshold_arcsec: float, margin_warning_fraction: float, + max_representative_radius_arcsec: float | None = None, ) -> None: """Validate parameters required by partition-local spatial deduplication.""" if crossmatch_radius_arcsec <= 0.0: @@ -123,6 +186,125 @@ def validate_spatial_safety( "margin_threshold_arcsec for partition-local deduplication " f"(radius={crossmatch_radius_arcsec}, margin={margin_threshold_arcsec})" ) + if max_representative_radius_arcsec is not None: + if max_representative_radius_arcsec < crossmatch_radius_arcsec: + raise ValueError( + "max_representative_radius_arcsec must be greater than or equal " + "to crossmatch_radius_arcsec" + ) + if max_representative_radius_arcsec >= margin_threshold_arcsec: + raise ValueError( + "max_representative_radius_arcsec must be smaller than " + "margin_threshold_arcsec" + ) + + +def _angular_distance_from_reference_arcsec( + ra_deg: np.ndarray, + dec_deg: np.ndarray, + reference_position: int, +) -> np.ndarray: + """Return great-circle distances from one array position in arcseconds.""" + ra = np.radians(ra_deg) + dec = np.radians(dec_deg) + rep_ra = ra[reference_position] + rep_dec = dec[reference_position] + dra = (ra - rep_ra + np.pi) % (2.0 * np.pi) - np.pi + hav = ( + np.sin((dec - rep_dec) / 2.0) ** 2 + + np.cos(dec) * np.cos(rep_dec) * np.sin(dra / 2.0) ** 2 + ) + return np.degrees(2.0 * np.arcsin(np.sqrt(np.clip(hav, 0.0, 1.0)))) * 3600.0 + + +def _split_groups_by_reference_radius( + df: pd.DataFrame, + initial_groups: pd.Series, + *, + max_radius_arcsec: float | None, + tiebreaking_priority: Sequence[str], + instrument_type_priority: Mapping[str, int] | None, + excluded_mask: pd.Series, + crd_col: str, +) -> tuple[pd.Series, pd.Series]: + """Greedily split components around deterministic best-ranked references.""" + if max_radius_arcsec is None or df.empty: + return ( + initial_groups.astype("int64"), + pd.Series(pd.NA, index=df.index, dtype="string"), + ) + + ranking = pd.DataFrame(index=df.index) + for order, column in enumerate(tiebreaking_priority): + if column == "instrument_type_homogenized": + if instrument_type_priority is None: + raise ValueError( + "instrument_type_priority is required when " + "'instrument_type_homogenized' is used in tiebreaking_priority." + ) + score = _score_instrument_type(df[column], instrument_type_priority) + elif column == "z_flag_homogenized": + score = _effective_z_flag_score(df) + else: + score = _to_numeric(df[column]) + ranking[f"__priority_{order}"] = score.astype("float64").fillna(-np.inf) + ranking["__crd"] = _canon_id_series(df[crd_col]).fillna("") + + sort_columns = [ + column for column in ranking.columns if column.startswith("__priority_") + ] + ["__crd"] + ascending = [False] * (len(sort_columns) - 1) + [True] + ordered_labels = ranking.sort_values( + sort_columns, + ascending=ascending, + kind="stable", + ).index + rank_position = pd.Series( + np.arange(len(ordered_labels), dtype="int64"), index=ordered_labels + ) + + ra = pd.to_numeric(df["ra"], errors="coerce").to_numpy(dtype="float64") + dec = pd.to_numeric(df["dec"], errors="coerce").to_numpy(dtype="float64") + group_values = initial_groups.to_numpy(dtype="int64", copy=True) + excluded = excluded_mask.to_numpy(dtype=bool, na_value=False) + result = np.full(len(df), -1, dtype="int64") + references = np.full(len(df), None, dtype=object) + next_group = 0 + + for group_value in pd.unique(group_values): + positions = np.flatnonzero((group_values == group_value) & ~excluded) + remaining = set(int(position) for position in positions) + while remaining: + reference = min( + remaining, + key=lambda position: int(rank_position.loc[df.index[position]]), + ) + candidates = np.fromiter(sorted(remaining), dtype="int64") + if np.isfinite(ra[reference]) and np.isfinite(dec[reference]): + candidate_ra = ra[candidates] + candidate_dec = dec[candidates] + local_reference = int(np.flatnonzero(candidates == reference)[0]) + distances = _angular_distance_from_reference_arcsec( + candidate_ra, + candidate_dec, + local_reference, + ) + selected = candidates[distances <= float(max_radius_arcsec) + 1e-9] + else: + selected = np.asarray([reference], dtype="int64") + result[selected] = next_group + references[selected] = str(df.iloc[reference][crd_col]).strip() + remaining.difference_update(int(position) for position in selected) + next_group += 1 + + for position in np.flatnonzero(excluded): + result[position] = next_group + references[position] = str(df.iloc[position][crd_col]).strip() + next_group += 1 + return ( + pd.Series(result, index=df.index, dtype="int64"), + pd.Series(references, index=df.index, dtype="string"), + ) def _assign_canonical_group_ids( @@ -140,24 +322,13 @@ def _validate_local_tie_invariants( *, group_col: str, tie_col: str, - z_flag_col: str = "z_flag_homogenized", ) -> None: """Validate winner/hard-tie semantics for every local component.""" - if z_flag_col not in df.columns: - raise KeyError( - f"Missing required semantic column '{z_flag_col}' for star classification" - ) - zf = pd.to_numeric(df[z_flag_col], errors="coerce") tie = pd.to_numeric(df[tie_col], errors="coerce") - stars = zf.eq(6.0) - if not tie[stars].eq(3).all(): - raise RuntimeError( - "Local tie invariant failed: every star must have tie_result=3" - ) - - nonstars = df.loc[~stars, [group_col]].copy() - nonstars["__tie"] = tie.loc[~stars].to_numpy() - for gid_value, component in nonstars.groupby(group_col, dropna=False): + participants = ~tie.eq(3.0) + participating = df.loc[participants, [group_col]].copy() + participating["__tie"] = tie.loc[participants].to_numpy() + for gid_value, component in participating.groupby(group_col, dropna=False): values = component["__tie"] n_one = int(values.eq(1).sum()) n_two = int(values.eq(2).sum()) @@ -194,14 +365,31 @@ def _log_representative_radius_diagnostics( if work.empty: return diagnostic - tie = pd.to_numeric(work[tie_col], errors="coerce") - work["_representative_rank"] = np.select([tie.eq(1), tie.eq(2)], [0, 1], default=2) - representatives = ( - work.sort_values([group_col, "_representative_rank", crd_col], kind="stable") - .drop_duplicates(group_col) - .set_index(group_col)[["ra", "dec"]] - .rename(columns={"ra": "_rep_ra", "dec": "_rep_dec"}) - ) + if SPLIT_REFERENCE_COLUMN in frame.columns: + work[SPLIT_REFERENCE_COLUMN] = frame.loc[ + work.index, SPLIT_REFERENCE_COLUMN + ].astype("string") + reference_rows = work[work[crd_col].astype("string").eq( + work[SPLIT_REFERENCE_COLUMN] + )] + representatives = ( + reference_rows.drop_duplicates(group_col) + .set_index(group_col)[["ra", "dec"]] + .rename(columns={"ra": "_rep_ra", "dec": "_rep_dec"}) + ) + else: + tie = pd.to_numeric(work[tie_col], errors="coerce") + work["_representative_rank"] = np.select( + [tie.eq(1), tie.eq(2)], [0, 1], default=2 + ) + representatives = ( + work.sort_values( + [group_col, "_representative_rank", crd_col], kind="stable" + ) + .drop_duplicates(group_col) + .set_index(group_col)[["ra", "dec"]] + .rename(columns={"ra": "_rep_ra", "dec": "_rep_dec"}) + ) work = work.join(representatives, on=group_col) ra = np.radians(work["ra"].to_numpy(dtype="float64")) @@ -399,11 +587,15 @@ def _edge_rows_partition( part: pd.DataFrame, crd_col: str, compared_col: str, - z_flag_col: str, + tie_col: str, ) -> pd.DataFrame: - """Extract directed non-star graph edges from one dataframe partition.""" - zf = pd.to_numeric(part[z_flag_col], errors="coerce") - work = part.loc[~zf.eq(6.0), [crd_col, compared_col]].copy() + """Extract directed graph edges from participating rows.""" + tie = ( + pd.to_numeric(part[tie_col], errors="coerce") + if tie_col in part.columns + else pd.Series(0, index=part.index, dtype="int8") + ) + work = part.loc[~tie.eq(3.0), [crd_col, compared_col]].copy() if work.empty: return pd.DataFrame( { @@ -429,10 +621,10 @@ def count_global_edge_group_mismatches( *, crd_col: str = "CRD_ID", compared_col: str = "compared_to", - z_flag_col: str = "z_flag_homogenized", + tie_col: str = "tie_result", group_col: str = "group_id", ): - """Return lazy counts of cross-group edges and dangling non-star edges.""" + """Return lazy counts of cross-group edges and dangling participating edges.""" meta = pd.DataFrame( { "u": pd.Series(dtype="string[pyarrow]"), @@ -443,14 +635,21 @@ def count_global_edge_group_mismatches( _edge_rows_partition, crd_col, compared_col, - z_flag_col, + tie_col, meta=meta, ).drop_duplicates() - zf = dd.to_numeric(df[z_flag_col], errors="coerce") + tie = ( + dd.to_numeric(df[tie_col], errors="coerce") + if tie_col in df.columns + else df[crd_col].map_partitions( + lambda s: pd.Series(0, index=s.index, dtype="int8"), + meta=pd.Series(dtype="int8"), + ) + ) groups = ( df[[crd_col, group_col]] - .assign(is_star=zf.eq(6.0)) + .assign(is_excluded=tie.eq(3.0)) .rename(columns={crd_col: "node", group_col: "node_group"}) ) groups = groups.assign(node=groups["node"].astype("string[pyarrow]")) @@ -460,17 +659,27 @@ def count_global_edge_group_mismatches( # can lose a left_on/right_on key while lowering consecutive merge/rename # expressions, producing a spurious merge key of None. groups_u = groups.rename( - columns={"node": "u", "node_group": "group_u", "is_star": "is_star_u"} + columns={ + "node": "u", + "node_group": "group_u", + "is_excluded": "is_excluded_u", + } ) groups_v = groups.rename( - columns={"node": "v", "node_group": "group_v", "is_star": "is_star_v"} + columns={ + "node": "v", + "node_group": "group_v", + "is_excluded": "is_excluded_v", + } ) checked = edges.merge(groups_u, on="u", how="left") checked = checked.merge(groups_v, on="v", how="left") dangling = checked["group_u"].isna() | checked["group_v"].isna() - both_nonstar = checked["is_star_u"].eq(False) & checked["is_star_v"].eq(False) - mismatch = (~dangling) & both_nonstar & checked["group_u"].ne(checked["group_v"]) + both_included = checked["is_excluded_u"].eq(False) & checked[ + "is_excluded_v" + ].eq(False) + mismatch = (~dangling) & both_included & checked["group_u"].ne(checked["group_v"]) return mismatch.sum(), dangling.sum() @@ -479,23 +688,38 @@ def count_global_tie_invariant_violations( *, group_col: str = "group_id", tie_col: str = "tie_result", - z_flag_col: str | None = "z_flag_homogenized", + z_flag_col: str | None = None, ): """Return a lazy count of groups violating final tie-result semantics.""" + invalid, missing_group_rows = build_global_tie_invariant_diagnostics( + df, + group_col=group_col, + tie_col=tie_col, + z_flag_col=z_flag_col, + ) + return invalid.map_partitions(len).sum() + missing_group_rows + + +def build_global_tie_invariant_diagnostics( + df: dd.DataFrame, + *, + group_col: str = "group_id", + tie_col: str = "tie_result", + z_flag_col: str | None = None, +) -> tuple[dd.DataFrame, object]: + """Build lazy per-group invariant diagnostics and missing-group count.""" tie = dd.to_numeric(df[tie_col], errors="coerce") - if z_flag_col is None: - nonstar_mask = ~tie.eq(3) - else: - zf = dd.to_numeric(df[z_flag_col], errors="coerce") - nonstar_mask = ~zf.eq(6.0) - nonstars = df.loc[nonstar_mask, [group_col]].assign( + # tie_result=3 is the public, type-agnostic marker for rows excluded from + # the graph. z_flag_col is retained only for API compatibility. + participating_mask = ~tie.eq(3) + participants = df.loc[participating_mask, [group_col]].assign( n=1, n0=tie.eq(0).astype("int8"), n1=tie.eq(1).astype("int8"), n2=tie.eq(2).astype("int8"), n_invalid=(~tie.isin([0, 1, 2])).astype("int8"), ) - stats = nonstars.groupby(group_col).agg( + stats = participants.groupby(group_col).agg( {"n": "sum", "n0": "sum", "n1": "sum", "n2": "sum", "n_invalid": "sum"} ) valid_single = ( @@ -505,8 +729,15 @@ def count_global_tie_invariant_violations( stats["n1"].eq(0) & stats["n2"].ge(2) & stats["n0"].eq(stats["n"] - stats["n2"]) ) invalid = stats["n_invalid"].gt(0) | ~(valid_single | valid_hard) - missing_group_rows = nonstars[group_col].isna().sum() - return invalid.sum() + missing_group_rows + missing_group_rows = participants[group_col].isna().sum() + diagnostics = stats.assign( + multiple_winners=stats["n1"].gt(1), + no_survivor=stats["n1"].eq(0) & stats["n2"].eq(0), + mixed_winner_hard_tie=stats["n1"].gt(0) & stats["n2"].gt(0), + single_hard_tie=stats["n1"].eq(0) & stats["n2"].eq(1), + invalid_tie_values=stats["n_invalid"].gt(0), + ) + return diagnostics.loc[invalid], missing_group_rows def _build_edges_fast( @@ -514,10 +745,10 @@ def _build_edges_fast( *, crd_col: str, compared_col: str, - zf_series: pd.Series | None = None, + excluded_mask: pd.Series | None = None, edge_log: bool = False, ): - """Build undirected edges among NON-STAR rows (vectorized path). + """Build undirected edges among rows participating in the graph. Returns: (nodes_index, edges_uv, diag) where: @@ -526,17 +757,16 @@ def _build_edges_fast( - diag: dict with basic diagnostics (counts). If `edge_log` is False, only cheap counts are filled; expensive ones are set to None. """ - # --- filter A-side (rows) to non-stars - non_star_mask = pd.Series(True, index=df.index) - if zf_series is not None: - non_star_mask &= ~pd.to_numeric(zf_series, errors="coerce").eq(6) + participating_mask = pd.Series(True, index=df.index, dtype="boolean") + if excluded_mask is not None: + participating_mask &= ~excluded_mask.reindex(df.index).fillna(False) - A_df = df.loc[non_star_mask, [crd_col, compared_col]].copy().reset_index(drop=True) + A_df = df.loc[participating_mask, [crd_col, compared_col]].copy().reset_index(drop=True) if A_df.empty: diag = { "edge_log_enabled": bool(edge_log), - "n_rows_nonstar": int(non_star_mask.sum()), - "n_rows_star_excluded": int((~non_star_mask).sum()), + "n_rows_nonstar": int(participating_mask.sum()), + "n_rows_star_excluded": int((~participating_mask).sum()), "n_edges_raw": 0, "n_edges_kept": 0, "n_edges_starB_excluded": None if not edge_log else 0, @@ -545,15 +775,13 @@ def _build_edges_fast( # Non-star IDs present on A-side present_nonstar = set( - _canon_id_series(df.loc[non_star_mask, crd_col]).dropna().unique() + _canon_id_series(df.loc[participating_mask, crd_col]).dropna().unique() ) # If diagnostics are enabled, precompute star IDs (for B-side exclusion count) - if edge_log and (zf_series is not None): + if edge_log and (excluded_mask is not None): star_ids = set( - _canon_id_series( - df.loc[pd.to_numeric(zf_series, errors="coerce").eq(6), crd_col] - ) + _canon_id_series(df.loc[~participating_mask, crd_col]) .dropna() .unique() ) @@ -589,8 +817,8 @@ def _build_edges_fast( if edges_raw.empty: diag = { "edge_log_enabled": bool(edge_log), - "n_rows_nonstar": int(non_star_mask.sum()), - "n_rows_star_excluded": int((~non_star_mask).sum()), + "n_rows_nonstar": int(participating_mask.sum()), + "n_rows_star_excluded": int((~participating_mask).sum()), "n_edges_raw": n_edges_raw, "n_edges_kept": 0, "n_edges_starB_excluded": n_edges_starB_excluded, @@ -603,18 +831,17 @@ def _build_edges_fast( ) # --- EXTRA LOG (edge_log): sanity-check that no star IDs leaked into edge nodes - if edge_log and (zf_series is not None) and len(nodes_edge): + if edge_log and (excluded_mask is not None) and len(nodes_edge): star_ids_fast = set( - _canon_id_series( - df.loc[pd.to_numeric(zf_series, errors="coerce").eq(6), crd_col] - ) + _canon_id_series(df.loc[~participating_mask, crd_col]) .dropna() .unique() ) leaked = [cid for cid in nodes_edge.astype("string") if cid in star_ids_fast] if leaked: _phase_logger().error( - "Star IDs leaked into fast-path edge nodes (logic violation): " + "Excluded object-type IDs leaked into fast-path edge nodes " + "(logic violation): " "count=%d sample=%s", len(leaked), leaked[:5], @@ -631,8 +858,8 @@ def _build_edges_fast( if lo.size == 0: diag = { "edge_log_enabled": bool(edge_log), - "n_rows_nonstar": int(non_star_mask.sum()), - "n_rows_star_excluded": int((~non_star_mask).sum()), + "n_rows_nonstar": int(participating_mask.sum()), + "n_rows_star_excluded": int((~participating_mask).sum()), "n_edges_raw": n_edges_raw, "n_edges_kept": 0, "n_edges_starB_excluded": n_edges_starB_excluded, @@ -645,8 +872,8 @@ def _build_edges_fast( diag = { "edge_log_enabled": bool(edge_log), - "n_rows_nonstar": int(non_star_mask.sum()), - "n_rows_star_excluded": int((~non_star_mask).sum()), + "n_rows_nonstar": int(participating_mask.sum()), + "n_rows_star_excluded": int((~participating_mask).sum()), "n_edges_raw": n_edges_raw, "n_edges_kept": int(uv.shape[0]), # None when edge_log=False to indicate we skipped the costly check @@ -861,14 +1088,16 @@ def _to_numeric(series_like) -> pd.Series: # ----------------------- # Guard restore # ----------------------- -def _only_star_neighbors_series(col: pd.Series, star_ids: set[str]) -> pd.Series: - """True when compared_to is non-empty AND all neighbors are star IDs.""" +def _only_excluded_neighbors_series( + col: pd.Series, excluded_ids: set[str] +) -> pd.Series: + """True when compared_to is non-empty and all neighbors are excluded.""" s = col.astype("string").fillna("") lst = s.str.split(",") out = [] for tokens in lst: toks = [t.strip() for t in tokens if t and t.strip()] - out.append(bool(toks) and all((tok in star_ids) for tok in toks)) + out.append(bool(toks) and all((tok in excluded_ids) for tok in toks)) return pd.Series(out, index=col.index, dtype="boolean") @@ -877,17 +1106,17 @@ def _apply_guard_restore_local( *, crd_col: str, compared_col: str, - zf_series: pd.Series | None, + excluded_mask: pd.Series, tie_col: str, tie_col_orig: str, ) -> pd.DataFrame: - """Restore original tie_result for non-stars with empty/only-star neighbors. + """Restore original labels for participants with no participating neighbors. Args: df: Partition dataframe. crd_col: Name of the CRD_ID column. compared_col: Name of the compared_to column. - zf_series: Optional z_flag series. + excluded_mask: Boolean mask for rows excluded from the graph. tie_col: Name of the tie_result column. tie_col_orig: Name of the original tie_result column. @@ -897,25 +1126,21 @@ def _apply_guard_restore_local( if tie_col_orig not in df.columns: return df # nothing to restore - # Stars (fixed as 3) and empty compared_to mask. - is_star = pd.Series(False, index=df.index) - if zf_series is not None: - is_star = pd.to_numeric(zf_series, errors="coerce").eq(6.0) + excluded = excluded_mask.reindex(df.index).fillna(False).astype(bool) cmp_str = df[compared_col].astype("string") cmp_empty = cmp_str.isna() | cmp_str.str.strip().eq("") - # Local set of star IDs for this partition/view. - star_ids = set(df.loc[is_star, crd_col].astype("string")) + excluded_ids = set(df.loc[excluded, crd_col].astype("string")) - only_star_neighbors = (~cmp_empty) & _only_star_neighbors_series( - df[compared_col], star_ids + only_excluded_neighbors = (~cmp_empty) & _only_excluded_neighbors_series( + df[compared_col], excluded_ids ) # Rule: - # - if star -> keep tie=3 (do not restore) - # - if non-star and (empty compared_to OR only star neighbors) -> restore - restore_mask = (~is_star) & (cmp_empty | only_star_neighbors) + # Excluded rows remain tie_result=3. Participating rows with no usable + # neighbors recover their original label. + restore_mask = (~excluded) & (cmp_empty | only_excluded_neighbors) # Apply restoration. df.loc[restore_mask, tie_col] = df.loc[restore_mask, tie_col_orig] @@ -933,19 +1158,21 @@ def _resolve_group( tiebreaking_priority: Sequence[str], instrument_type_priority: Mapping[str, int] | None, delta_z_threshold: float, + excluded_mask: pd.Series | None = None, ) -> pd.DataFrame: """Resolve ties within a single connected component.""" crd = crd_col out = g[[crd]].copy() out["tie_result_new"] = 0 - star_mask = pd.Series(False, index=g.index) - if "z_flag_homogenized" in g.columns: - zf_series = _to_numeric(g["z_flag_homogenized"]) - star_mask = zf_series.eq(6) - out.loc[star_mask.index[star_mask], "tie_result_new"] = 3 + excluded = ( + pd.Series(False, index=g.index) + if excluded_mask is None + else excluded_mask.reindex(g.index).fillna(False).astype(bool) + ) + out.loc[excluded.index[excluded], "tie_result_new"] = 3 - cand = g[~star_mask].copy() + cand = g[~excluded].copy() if cand.empty: return out[[crd, "tie_result_new"]] @@ -1012,6 +1239,9 @@ def deduplicate_pandas( partition_tag: str | None = None, logger: logging.LoggerAdapter | None = None, group_col: str | None = None, # new + object_type_inclusion: Mapping[str, object] | None = None, + max_representative_radius_arcsec: float | None = None, + preserve_split_reference: bool = False, ) -> pd.DataFrame: """Graph-based deduplication with vectorized per-group resolution and Dz collapse. @@ -1038,6 +1268,19 @@ def deduplicate_pandas( raise KeyError(f"Missing required columns: {missing}") out = df.copy() + inclusion = validate_object_type_inclusion(object_type_inclusion) + excluded_object_types = sorted( + object_type + for object_type, option in _OBJECT_TYPE_TO_INCLUDE_KEY.items() + if not inclusion[option] + ) + if not inclusion["include_unclassified"]: + excluded_object_types.append("unclassified") + object_types = out.get( + "object_type_homogenized", + pd.Series(pd.NA, index=out.index, dtype="string"), + ) + excluded_mask = _excluded_object_type_mask(object_types, inclusion) tie_col_orig = f"{tie_col}_orig" if tie_col in out.columns: @@ -1049,16 +1292,12 @@ def deduplicate_pandas( crd_norm = out[crd_col].astype("string").str.strip() priority_set = set(tiebreaking_priority) - zf_series: pd.Series | None = None - if "z_flag_homogenized" in out.columns: - zf_series = _to_numeric(out["z_flag_homogenized"]) - # Pass edge_log down so diagnostics are computed only when requested. nodes_edge, edges_uv, diag = _build_edges_fast( out, crd_col=crd_col, compared_col=compared_col, - zf_series=zf_series, + excluded_mask=excluded_mask, edge_log=edge_log, ) @@ -1066,23 +1305,29 @@ def deduplicate_pandas( if edge_log: lg = logger or _phase_logger() tag = f"[{partition_tag}]" if partition_tag else "[global]" - starB = diag.get("n_edges_starB_excluded") - if isinstance(starB, int) and starB > 0: + excluded_neighbor_edges = diag.get("n_edges_starB_excluded") + if isinstance(excluded_neighbor_edges, int) and excluded_neighbor_edges > 0: lg.warning( - "%s Star neighbors excluded during edge build: star_rows_excl=%d, edges_raw=%d, starB_excl=%d, edges_kept=%d", + "%s Object types excluded from deduplication graph: " + "types=%s, rows_excluded=%d, edges_raw=%d, " + "excluded_neighbor_edges=%d, edges_kept=%d", tag, + excluded_object_types, diag.get("n_rows_star_excluded", 0), diag.get("n_edges_raw", 0), - starB, + excluded_neighbor_edges, diag.get("n_edges_kept", 0), ) elif partition_tag is None: lg.info( - "%s Edge build summary: star_rows_excl=%d, edges_raw=%d, starB_excl=%s, edges_kept=%d", + "%s Edge build summary: excluded_object_types=%s, " + "rows_excluded=%d, edges_raw=%d, " + "excluded_neighbor_edges=%s, edges_kept=%d", tag, + excluded_object_types, diag.get("n_rows_star_excluded", 0), diag.get("n_edges_raw", 0), - str(starB), + str(excluded_neighbor_edges), diag.get("n_edges_kept", 0), ) @@ -1108,13 +1353,11 @@ def deduplicate_pandas( # Try to bridge NA rows via neighbor groups from the fast path label map if na_mask.any() and labels_edge.size: pos_na = np.flatnonzero(na_mask) - # Stars never participate in non-star components, even when they point - # to an already-labelled neighbor. Leave them for the singleton path. - if zf_series is not None: - bridge_is_star = np.asarray( - zf_series.iloc[pos_na].eq(6).fillna(False), dtype=bool - ) - pos_na = pos_na[~bridge_is_star] + # Excluded rows never bridge into participating components. + bridge_is_excluded = excluded_mask.iloc[pos_na].fillna(False).to_numpy( + dtype=bool + ) + pos_na = pos_na[~bridge_is_excluded] cmp_str_all = out[compared_col].astype("string").str.strip() cmp_lists = cmp_str_all.iloc[pos_na].str.split(",") sub = pd.DataFrame({"pos": pos_na, "nbr": cmp_lists}).explode( @@ -1138,26 +1381,20 @@ def deduplicate_pandas( if na_mask.any(): pos_na = np.flatnonzero(na_mask) - # Stars within NA. - if zf_series is not None: - is_star_na = np.asarray( - zf_series.iloc[pos_na].eq(6).fillna(False), dtype=bool - ) - else: - is_star_na = np.zeros(pos_na.size, dtype=bool) + is_excluded_na = excluded_mask.iloc[pos_na].fillna(False).to_numpy(dtype=bool) - pos_na_nonstar = pos_na[~is_star_na] + pos_na_participating = pos_na[~is_excluded_na] # Next free ID, compatible with fast-path labels. next_gid = int(labels_edge.max()) + 1 if labels_edge.size else 0 - # NA non-star rows. - if pos_na_nonstar.size: + # Participating rows without a fast-path group. + if pos_na_participating.size: crd_arr_all = crd_norm.to_numpy() cmp_arr_all = out[compared_col].astype("string").str.strip().to_numpy() - crd_arr = crd_arr_all[pos_na_nonstar] - cmp_arr = cmp_arr_all[pos_na_nonstar] + crd_arr = crd_arr_all[pos_na_participating] + cmp_arr = cmp_arr_all[pos_na_participating] sub_df = pd.DataFrame({crd_col: crd_arr, compared_col: cmp_arr}) edges_na = _build_edges_pdf( @@ -1168,14 +1405,14 @@ def deduplicate_pandas( if nodes_na: if edges_na.empty: # No edges: each row becomes a singleton. - gids[pos_na_nonstar] = np.arange( - next_gid, next_gid + len(pos_na_nonstar), dtype=np.int64 + gids[pos_na_participating] = np.arange( + next_gid, next_gid + len(pos_na_participating), dtype=np.int64 ) - next_gid += len(pos_na_nonstar) + next_gid += len(pos_na_participating) else: # With edges: real components. comp_map = _connected_components(nodes_na, edges_na) - gids[pos_na_nonstar] = next_gid + np.fromiter( + gids[pos_na_participating] = next_gid + np.fromiter( (comp_map.get(str(cid), -1) for cid in crd_arr), dtype=np.int64, count=len(crd_arr), @@ -1183,49 +1420,50 @@ def deduplicate_pandas( next_gid += max(comp_map.values()) + 1 if comp_map else 0 else: # No nodes: singleton per row. - gids[pos_na_nonstar] = np.arange( - next_gid, next_gid + len(pos_na_nonstar), dtype=np.int64 + gids[pos_na_participating] = np.arange( + next_gid, next_gid + len(pos_na_participating), dtype=np.int64 ) - next_gid += len(pos_na_nonstar) + next_gid += len(pos_na_participating) # Mark as mapped to avoid later collisions. - na_mask[pos_na_nonstar] = False + na_mask[pos_na_participating] = False - # Stars (each is a singleton). - if is_star_na.any(): - n_star = int(is_star_na.sum()) - gids[pos_na[is_star_na]] = np.arange( - next_gid, next_gid + n_star, dtype=np.int64 + # Excluded objects are isolated singletons. + if is_excluded_na.any(): + n_excluded = int(is_excluded_na.sum()) + gids[pos_na[is_excluded_na]] = np.arange( + next_gid, next_gid + n_excluded, dtype=np.int64 ) - na_mask[pos_na[is_star_na]] = False - next_gid += n_star + na_mask[pos_na[is_excluded_na]] = False + next_gid += n_excluded - out["__group__"] = gids + out["__group__"], out[SPLIT_REFERENCE_COLUMN] = _split_groups_by_reference_radius( + out, + pd.Series(gids, index=out.index), + max_radius_arcsec=max_representative_radius_arcsec, + tiebreaking_priority=tiebreaking_priority, + instrument_type_priority=instrument_type_priority, + excluded_mask=excluded_mask, + crd_col=crd_col, + ) gid = out["__group__"] crd_s = crd_norm group_sizes = pd.Series(1, index=out.index).groupby(gid).transform("sum") is_singleton = group_sizes.eq(1) - if zf_series is None: - is_star = pd.Series(False, index=out.index) - else: - is_star = zf_series.eq(6) - is_singleton_np = is_singleton.to_numpy(dtype=bool, na_value=False) - is_star_np = is_star.to_numpy(dtype=bool, na_value=False) + is_excluded_np = excluded_mask.to_numpy(dtype=bool, na_value=False) tr = np.zeros(len(out), dtype=np.int8) - tr[is_star_np] = 3 - tr[is_singleton_np & ~is_star_np] = 1 + tr[is_excluded_np] = 3 + tr[is_singleton_np & ~is_excluded_np] = 1 is_multi = ~is_singleton - non_star = ~is_star - survivors = (is_multi & non_star).copy() + participating = ~excluded_mask + survivors = (is_multi & participating).copy() - zf_num = _to_numeric( - out.get("z_flag_homogenized", pd.Series(np.nan, index=out.index)) - ).astype("float64") + zf_num = _effective_z_flag_score(out) if "instrument_type_homogenized" in priority_set: if instrument_type_priority is None: @@ -1269,18 +1507,17 @@ def deduplicate_pandas( out[tie_col] = pd.Series(tr, index=out.index).astype("Int8") - if zf_series is not None: - tr_num = pd.to_numeric(out[tie_col], errors="coerce") - eq3_np = tr_num.eq(3.0).to_numpy(dtype=bool, na_value=False) - is_star_np = is_star.to_numpy(dtype=bool, na_value=False) - is_single_np = is_singleton.to_numpy(dtype=bool, na_value=False) + tr_num = pd.to_numeric(out[tie_col], errors="coerce") + eq3_np = tr_num.eq(3.0).to_numpy(dtype=bool, na_value=False) + excluded_np = excluded_mask.to_numpy(dtype=bool, na_value=False) + is_single_np = is_singleton.to_numpy(dtype=bool, na_value=False) - invalid_3_np = eq3_np & ~is_star_np - if invalid_3_np.any(): - invalid_3 = pd.Series(invalid_3_np, index=out.index) - single = pd.Series(is_single_np, index=out.index) - out.loc[invalid_3 & single, tie_col] = np.int8(1) - out.loc[invalid_3 & ~single, tie_col] = np.int8(0) + invalid_3_np = eq3_np & ~excluded_np + if invalid_3_np.any(): + invalid_3 = pd.Series(invalid_3_np, index=out.index) + single = pd.Series(is_single_np, index=out.index) + out.loc[invalid_3 & single, tie_col] = np.int8(1) + out.loc[invalid_3 & ~single, tie_col] = np.int8(0) z_num = _to_numeric(out[z_col]).astype("float64") f_num = zf_num.fillna(-np.inf) @@ -1414,7 +1651,7 @@ def deduplicate_pandas( out, crd_col=crd_col, compared_col=compared_col, - zf_series=zf_series, + excluded_mask=excluded_mask, tie_col=tie_col, tie_col_orig=tie_col_orig, ) @@ -1428,17 +1665,15 @@ def deduplicate_pandas( else: drop_cols.append("__group__") + if not preserve_split_reference: + drop_cols.append(SPLIT_REFERENCE_COLUMN) + out.drop(columns=drop_cols, inplace=True, errors="ignore") - if zf_series is not None: - out.loc[zf_series.eq(6).fillna(False), tie_col] = np.int8(3) + out.loc[excluded_mask, tie_col] = np.int8(3) if group_col: - _validate_local_tie_invariants( - out, - group_col=group_col, - tie_col=tie_col, - ) + _validate_local_tie_invariants(out, group_col=group_col, tie_col=tie_col) return out @@ -1521,6 +1756,8 @@ def _dedup_local_with_margin( margin_threshold_arcsec: float = 5.0, margin_warning_fraction: float = 0.8, representative_radius_diagnostics_enabled: bool = False, + object_type_inclusion: Mapping[str, object] | None = None, + max_representative_radius_arcsec: float | None = None, ) -> pd.DataFrame: """Run dedup on (main + margin) and return labels for main rows only. @@ -1554,6 +1791,7 @@ def _dedup_local_with_margin( "ra", "dec", "z_flag_homogenized", + "object_type_homogenized", } | set( tiebreaking_priority or [] ) @@ -1620,6 +1858,9 @@ def _dedup_local_with_margin( partition_tag=partition_tag, logger=_phase_logger(), group_col=group_col, + object_type_inclusion=object_type_inclusion, + max_representative_radius_arcsec=max_representative_radius_arcsec, + preserve_split_reference=representative_radius_diagnostics_enabled, ) if representative_radius_diagnostics_enabled: solved[REPRESENTATIVE_RADIUS_DIAGNOSTIC_COLUMN] = ( @@ -1628,31 +1869,40 @@ def _dedup_local_with_margin( group_col=group_col, tie_col=tie_col, crd_col=crd_col, - radius_arcsec=crossmatch_radius_arcsec, + radius_arcsec=( + max_representative_radius_arcsec + if max_representative_radius_arcsec is not None + else crossmatch_radius_arcsec + ), partition_tag=partition_tag, ) ) if group_col and group_col in solved.columns: - # Every non-star edge whose endpoints are present in this local view must - # resolve to one canonical component. This catches graph/label drift at - # the partition boundary before labels are merged globally. - zf = pd.to_numeric(solved.get("z_flag_homogenized"), errors="coerce") - edge_nodes, edge_uv, _ = _build_edges_fast( - solved, - crd_col=crd_col, - compared_col=compared_col, - zf_series=zf, - edge_log=False, - ) - if edge_uv.size: - group_by_id = solved.drop_duplicates(crd_col).set_index(crd_col)[group_col] - left = group_by_id.reindex(edge_nodes.take(edge_uv[:, 0])).to_numpy() - right = group_by_id.reindex(edge_nodes.take(edge_uv[:, 1])).to_numpy() - if np.any(left != right): - raise RuntimeError( - f"{partition_tag}: non-star edge endpoints received different group_id values" - ) + if max_representative_radius_arcsec is None: + # Without radius truncation, every participating edge must remain + # inside one canonical component. + semantic_exclusion = pd.to_numeric( + solved[tie_col], errors="coerce" + ).eq(3.0) + edge_nodes, edge_uv, _ = _build_edges_fast( + solved, + crd_col=crd_col, + compared_col=compared_col, + excluded_mask=semantic_exclusion, + edge_log=False, + ) + if edge_uv.size: + group_by_id = solved.drop_duplicates(crd_col).set_index(crd_col)[ + group_col + ] + left = group_by_id.reindex(edge_nodes.take(edge_uv[:, 0])).to_numpy() + right = group_by_id.reindex(edge_nodes.take(edge_uv[:, 1])).to_numpy() + if np.any(left != right): + raise RuntimeError( + f"{partition_tag}: participating edge endpoints received " + "different group_id values" + ) if representative_radius_diagnostics_enabled: src_counts = solved.groupby(group_col)["_src"].nunique() @@ -1739,6 +1989,8 @@ def _dedup_alignfunc_with_margin( margin_threshold_arcsec: float = 5.0, margin_warning_fraction: float = 0.8, representative_radius_diagnostics_enabled: bool = False, + object_type_inclusion: Mapping[str, object] | None = None, + max_representative_radius_arcsec: float | None = None, ) -> pd.DataFrame: """Adapter for LSDB/HATS `align_and_apply`. @@ -1770,6 +2022,8 @@ def _dedup_alignfunc_with_margin( margin_threshold_arcsec=margin_threshold_arcsec, margin_warning_fraction=margin_warning_fraction, representative_radius_diagnostics_enabled=representative_radius_diagnostics_enabled, + object_type_inclusion=object_type_inclusion, + max_representative_radius_arcsec=max_representative_radius_arcsec, ) @@ -1787,6 +2041,8 @@ def _dedup_local_no_margin( group_col: str | None = None, crossmatch_radius_arcsec: float = 0.5, representative_radius_diagnostics_enabled: bool = False, + object_type_inclusion: Mapping[str, object] | None = None, + max_representative_radius_arcsec: float | None = None, ) -> pd.DataFrame: """Run dedup using only the main partition. @@ -1817,6 +2073,7 @@ def _dedup_local_no_margin( "ra", "dec", "z_flag_homogenized", + "object_type_homogenized", } | set( tiebreaking_priority or [] ) @@ -1857,6 +2114,9 @@ def _dedup_local_no_margin( partition_tag=partition_tag, logger=_phase_logger(), group_col=group_col, + object_type_inclusion=object_type_inclusion, + max_representative_radius_arcsec=max_representative_radius_arcsec, + preserve_split_reference=representative_radius_diagnostics_enabled, ) if representative_radius_diagnostics_enabled: solved[REPRESENTATIVE_RADIUS_DIAGNOSTIC_COLUMN] = ( @@ -1865,7 +2125,11 @@ def _dedup_local_no_margin( group_col=group_col, tie_col=tie_col, crd_col=crd_col, - radius_arcsec=crossmatch_radius_arcsec, + radius_arcsec=( + max_representative_radius_arcsec + if max_representative_radius_arcsec is not None + else crossmatch_radius_arcsec + ), partition_tag=partition_tag, ) ) @@ -1929,6 +2193,8 @@ def run_dedup_with_lsdb_map_partitions( margin_threshold_arcsec: float = 5.0, margin_warning_fraction: float = 0.8, representative_radius_diagnostics_enabled: bool = False, + object_type_inclusion: Mapping[str, object] | None = None, + max_representative_radius_arcsec: float | None = None, ) -> dd.DataFrame: """Compute dedup labels per partition via LSDB; align divisions if margin exists. @@ -1975,6 +2241,7 @@ def run_dedup_with_lsdb_map_partitions( compared_col, z_col, "z_flag_homogenized", + "object_type_homogenized", } _assert_required(main_ddf, required_base, "main") _assert_priorities(main_ddf, list(tiebreaking_priority), "main") @@ -2022,6 +2289,8 @@ def run_dedup_with_lsdb_map_partitions( group_col=group_col, crossmatch_radius_arcsec=crossmatch_radius_arcsec, representative_radius_diagnostics_enabled=representative_radius_diagnostics_enabled, + object_type_inclusion=object_type_inclusion, + max_representative_radius_arcsec=max_representative_radius_arcsec, ) else: # ------------------------------------------------------------------ @@ -2088,6 +2357,8 @@ def run_dedup_with_lsdb_map_partitions( margin_threshold_arcsec=float(margin_threshold_arcsec), margin_warning_fraction=float(margin_warning_fraction), representative_radius_diagnostics_enabled=representative_radius_diagnostics_enabled, + object_type_inclusion=object_type_inclusion, + max_representative_radius_arcsec=max_representative_radius_arcsec, ) # Build a Dask DataFrame from the delayed per-pixel label frames. diff --git a/packages/specz.py b/packages/specz.py index c38acce..c4dd001 100644 --- a/packages/specz.py +++ b/packages/specz.py @@ -1340,6 +1340,7 @@ def _validate_and_rename( "z": DTYPE_FLOAT, "z_flag": DTYPE_FLOAT, "z_err": DTYPE_FLOAT, + "object_type": DTYPE_STR, } for col, pd_dtype in base_schema.items(): if col not in df.columns: @@ -1626,6 +1627,7 @@ def _normalize_extra_columns_config(value: Any) -> dict[str, dict[str, str]]: "group_id", "z_flag_homogenized", "instrument_type_homogenized", + "object_type_homogenized", "is_in_DP1_fields", "is_in_rubin_footprint", } @@ -1637,6 +1639,45 @@ def _normalize_extra_columns_config(value: Any) -> dict[str, dict[str, str]]: return normalized +def build_runtime_schema_hints( + param_config: dict, translation_config: dict +) -> dict[str, str]: + """Build schema hints that must survive every crossmatch round. + + Args: + param_config: Pipeline ``param`` configuration. + translation_config: Validated flags translation configuration. + + Returns: + Mapping of output column names to ``str``, ``float``, ``int`` or ``bool``. + """ + hints: dict[str, str] = {} + if bool(translation_config.get("save_expr_columns", False)): + hints.update( + _normalize_schema_hints(translation_config.get("expr_column_schema")) + ) + + extra_columns = _normalize_extra_columns_config(param_config.get("extra_columns")) + hints.update({output: spec["type"] for output, spec in extra_columns.items()}) + + if _as_bool_config(param_config.get("insert_DP1_footprint_flag"), default=False): + hints["is_in_DP1_fields"] = "int" + if _as_bool_config( + param_config.get("insert_rubin_footprint_flag"), default=False + ): + hints["is_in_rubin_footprint"] = "int" + + standard_priorities = { + "z_flag_homogenized", + "instrument_type_homogenized", + } + for column in translation_config.get("tiebreaking_priority", []) or []: + name = str(column).strip() + if name and name not in standard_priorities: + hints[name] = "float" + return hints + + def _copy_extra_columns_from_sources( df: dd.DataFrame, columns: dict[str, dict[str, str]], @@ -2180,7 +2221,10 @@ def _select_output_columns( "group_id", ] - # Optional homogenized fields. + # Homogenized object type is part of every output, including all-null catalogs. + final_cols.append("object_type_homogenized") + + # Optional homogenized fields used by the legacy quality/ranking logic. if "z_flag_homogenized" in df.columns: final_cols.append("z_flag_homogenized") if "instrument_type_homogenized" in df.columns: @@ -2344,7 +2388,7 @@ def _requires_z_flag_homogenization(combine_mode: str, cut_value: object) -> boo numeric_cut = float(cut_value) except (TypeError, ValueError): return False - return numeric_cut in {1.0, 2.0, 3.0, 4.0, 5.0, 6.0} + return numeric_cut in {1.0, 2.0, 3.0, 4.0} def validate_combine_configuration( @@ -2381,20 +2425,6 @@ def validate_combine_configuration( f"tiebreaking_priority must be non-empty for {normalized_mode}" ) - try: - numeric_cut = float(cut_value) if cut_value is not None else None - except (TypeError, ValueError): - numeric_cut = None - if ( - normalized_mode == "concatenate_and_remove_duplicates" - and numeric_cut == 6.0 - and logger is not None - ): - logger.warning( - "z_flag_homogenized_value_to_cut=6 retains only stars, while " - "concatenate_and_remove_duplicates excludes tie_result=3; the final " - "catalog will normally be empty." - ) return normalized_mode, priorities @@ -2561,10 +2591,10 @@ def prepare_catalog( # Zero is the explicit no-cut sentinel. It is summarized once # by the driver instead of repeated for every input catalog. pass - elif cut_val not in {1.0, 2.0, 3.0, 4.0, 5.0, 6.0}: + elif cut_val not in {1.0, 2.0, 3.0, 4.0}: lg.warning( "Invalid z_flag_homogenized_value_to_cut=%s; use 0 to disable " - "the cut or one of 1, 2, 3, 4, 5, 6. Skipping cut.", + "the cut or one of 1, 2, 3, 4. Skipping cut.", z_flag_homogenized_value_to_cut, ) else: @@ -2646,7 +2676,7 @@ def prepare_catalog( param_config.get("insert_DP1_footprint_flag", False), default=False ) insert_rubin = _as_bool_config( - param_config.get("insert_rubin_footprint_flag", True), default=True + param_config.get("insert_rubin_footprint_flag", False), default=False ) if insert_dp1: diff --git a/packages/specz_homogenization.py b/packages/specz_homogenization.py index 1c2711a..94994d9 100644 --- a/packages/specz_homogenization.py +++ b/packages/specz_homogenization.py @@ -6,6 +6,7 @@ homogenized columns: - z_flag_homogenized - instrument_type_homogenized + - object_type_homogenized It is intended to be imported by the main `specz.py` module. Functions are copied verbatim from the original file to avoid behavior changes. @@ -20,6 +21,7 @@ # ----------------------- import ast as _ast import builtins +import difflib import logging import math @@ -66,6 +68,232 @@ "LRB_A": 4.0, "MR_A": 4.0, } +_TRANSLATION_KEYS = { + "z_flag_translation": ("float", {0.0, 1.0, 2.0, 3.0, 4.0}), + "instrument_type_translation": ("str", {"s", "g", "p"}), + "object_type_translation": ( + "str", + {"star", "galactic", "qso", "agn", "galaxy"}, + ), +} +_RULE_OPTIONS = { + "conditions", "default", "source", "optional_source", + "allow_condition_overlap", "fast_path", +} +_TOP_LEVEL_KEYS = { + "tiebreaking_priority", "delta_z_threshold", "crossmatch_radius_arcsec", + "max_representative_radius_arcsec", + "margin_threshold_arcsec", "margin_warning_fraction", + "validate_global_graph_edges", "validate_global_tie_invariants", + "validate_crd_id_uniqueness", "repartition_prepared_catalogs", + "prepared_partition_size", "crossmatch_n_neighbors", + "crossmatch_saturation_enabled", "crossmatch_saturation_warn_fraction", + "crossmatch_saturation_fail_fraction", + "crossmatch_geometry_diagnostics_enabled", + "representative_radius_diagnostics_enabled", "dedup_edge_diagnostics_enabled", + "tie_invariant_diagnostics_enabled", + "tie_invariant_diagnostics_detailed_enabled", + "tie_invariant_diagnostics_sample_size", + "tie_invariant_diagnostics_max_rows", + "label_merge_diagnostics_enabled", + "instrument_type_priority", "save_expr_columns", "expr_column_schema", + "runtime_schema_hints", "translation_rules", +} +_SAFE_FUNCTIONS = {"len", "int", "float", "str"} +_SAFE_NUMPY_FUNCTIONS = {"isfinite", "abs", "trunc"} +_SAFE_SERIES_METHODS = {"isin", "contains", "strip", "lower"} +_SAFE_AST_NODES = ( + _ast.Expression, _ast.BoolOp, _ast.BinOp, _ast.UnaryOp, _ast.Compare, + _ast.Name, _ast.Load, _ast.Constant, _ast.Call, _ast.Attribute, + _ast.Subscript, _ast.Slice, _ast.List, _ast.Tuple, _ast.keyword, + _ast.And, _ast.Or, _ast.Not, _ast.Invert, _ast.UAdd, _ast.USub, + _ast.Add, _ast.Sub, _ast.Mult, _ast.Div, _ast.FloorDiv, _ast.Mod, + _ast.Pow, _ast.BitAnd, _ast.BitOr, _ast.Eq, _ast.NotEq, _ast.Lt, + _ast.LtE, _ast.Gt, _ast.GtE, _ast.In, _ast.NotIn, +) + + +class _SafeExpressionValidator(_ast.NodeVisitor): + """Reject expression syntax outside the small vectorized-rule DSL.""" + + def __init__(self, path: str): + self.path = path + + def fail(self, node: _ast.AST, detail: str) -> None: + raise ValueError( + f"{self.path}: forbidden expression element {type(node).__name__}: {detail}" + ) + + def generic_visit(self, node: _ast.AST) -> None: + if not isinstance(node, _SAFE_AST_NODES): + self.fail(node, "not allowed by the translation expression DSL") + super().generic_visit(node) + + def visit_Name(self, node: _ast.Name) -> None: + if node.id.startswith("_"): + self.fail(node, f"private name {node.id!r} is not allowed") + + def visit_Attribute(self, node: _ast.Attribute) -> None: + if node.attr.startswith("_"): + self.fail(node, f"private attribute {node.attr!r} is not allowed") + if isinstance(node.value, _ast.Name) and node.value.id == "np": + if node.attr not in _SAFE_NUMPY_FUNCTIONS: + self.fail(node, f"np.{node.attr} is not allowed") + elif node.attr not in _SAFE_SERIES_METHODS and node.attr != "str": + self.fail(node, f"attribute or method {node.attr!r} is not allowed") + self.generic_visit(node) + + def visit_Call(self, node: _ast.Call) -> None: + if isinstance(node.func, _ast.Name): + if node.func.id not in _SAFE_FUNCTIONS: + self.fail(node, f"function {node.func.id!r} is not allowed") + elif isinstance(node.func, _ast.Attribute): + self.visit_Attribute(node.func) + else: + self.fail(node, "only whitelisted named functions and methods are allowed") + for arg in node.args: + self.visit(arg) + for keyword in node.keywords: + if keyword.arg is None or keyword.arg.startswith("_"): + self.fail(keyword, "keyword expansion/private keywords are not allowed") + self.visit(keyword.value) + + +def _validate_expression(expr: str, path: str) -> None: + try: + tree = _ast.parse(expr, mode="eval") + except SyntaxError as exc: + raise ValueError(f"{path}: invalid expression syntax: {exc.msg}") from exc + _SafeExpressionValidator(path).visit(tree) + + +def validate_translation_config(config: dict) -> None: + """Validate user-editable translation rules before catalog processing.""" + if not isinstance(config, dict): + raise TypeError("flags translation root must be a mapping") + unknown_top = set(config) - _TOP_LEVEL_KEYS + if unknown_top: + raise ValueError(f"flags translation root: unknown option(s): {sorted(unknown_top)}") + for key in ( + "tie_invariant_diagnostics_enabled", + "tie_invariant_diagnostics_detailed_enabled", + "label_merge_diagnostics_enabled", + ): + if key in config and not isinstance(config[key], bool): + raise TypeError(f"{key} must be a boolean") + for key in ( + "tie_invariant_diagnostics_sample_size", + "tie_invariant_diagnostics_max_rows", + ): + if key in config and ( + not isinstance(config[key], int) + or isinstance(config[key], bool) + or config[key] < 1 + ): + raise ValueError(f"{key} must be a positive integer") + max_radius = config.get("max_representative_radius_arcsec") + if max_radius is not None and ( + isinstance(max_radius, bool) + or not isinstance(max_radius, (int, float)) + or float(max_radius) <= 0.0 + ): + raise ValueError( + "max_representative_radius_arcsec must be a positive number or null" + ) + rules = config.get("translation_rules", {}) + runtime_hints = config.get("runtime_schema_hints", {}) + if not isinstance(runtime_hints, dict): + raise TypeError("runtime_schema_hints must be a mapping") + for column, kind in runtime_hints.items(): + if not isinstance(column, str) or not column.strip(): + raise ValueError("runtime_schema_hints column names must be non-empty strings") + if kind not in {"str", "float", "int", "bool"}: + raise ValueError( + f"runtime_schema_hints.{column}={kind!r} is invalid; " + "expected str, float, int or bool" + ) + if not isinstance(rules, dict): + raise TypeError("translation_rules must be a mapping") + for survey, ruleset in rules.items(): + base = f"translation_rules.{survey}" + if not isinstance(survey, str) or not survey.strip(): + raise ValueError("translation_rules survey names must be non-empty strings") + if not isinstance(ruleset, dict): + raise TypeError(f"{base} must be a mapping") + unknown_blocks = set(ruleset) - set(_TRANSLATION_KEYS) + if unknown_blocks: + raise ValueError(f"{base}: unknown option(s): {sorted(unknown_blocks)}") + for block, (kind, allowed) in _TRANSLATION_KEYS.items(): + if block not in ruleset: + continue + path = f"{base}.{block}" + rule = ruleset[block] + if not isinstance(rule, dict): + raise TypeError(f"{path} must be a mapping") + for option in ("optional_source", "allow_condition_overlap"): + if option in rule and not isinstance(rule[option], bool): + raise TypeError(f"{path}.{option} must be a boolean") + if "source" in rule and ( + not isinstance(rule["source"], str) or not rule["source"].strip() + ): + raise ValueError(f"{path}.source must be a non-empty string") + policy = rule.get("fast_path", "auto") + if policy not in {"auto", "disabled", "required"}: + raise ValueError( + f"{path}.fast_path must be one of ['auto', 'disabled', 'required']" + ) + conditions = rule.get("conditions", []) + if not isinstance(conditions, list): + raise TypeError(f"{path}.conditions must be a list") + for index, condition in enumerate(conditions): + cpath = f"{path}.conditions[{index}]" + if not isinstance(condition, dict): + raise TypeError(f"{cpath} must be a mapping") + unknown = set(condition) - {"expr", "value"} + if unknown: + raise ValueError(f"{cpath}: unknown option(s): {sorted(unknown)}") + if not isinstance(condition.get("expr"), str) or not condition["expr"].strip(): + raise ValueError(f"{cpath}.expr must be a non-empty string") + if "value" not in condition: + raise ValueError(f"{cpath}.value is required") + _validate_expression(condition["expr"], f"{cpath}.expr") + values = [] + if "default" in rule: + values.append((f"{path}.default", rule["default"])) + values.extend( + (f"{path}.conditions[{i}].value", c["value"]) + for i, c in enumerate(conditions) + ) + for key, value in rule.items(): + if key not in _RULE_OPTIONS: + close = difflib.get_close_matches(str(key), _RULE_OPTIONS, n=1, cutoff=0.8) + if close: + raise ValueError( + f"{path}.{key}: unknown option; did you mean {close[0]!r}?" + ) + values.append((f"{path}.{key}", value)) + if block == "z_flag_translation": + try: + float(key) + except (TypeError, ValueError) as exc: + raise ValueError( + f"{path}.{key}: z_flag direct mapping keys must be numeric" + ) from exc + for value_path, value in values: + if value is None: + continue + normalized = str(value).strip().lower() if kind == "str" else value + if kind == "float": + try: + normalized = float(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{value_path} must be numeric or null") from exc + if normalized not in allowed: + raise ValueError( + f"{value_path}={value!r} is invalid; allowed values are " + f"{sorted(allowed)} or null" + ) + # ----------------------- # Local helper (duplicated to avoid circular dep) # ----------------------- @@ -129,6 +357,17 @@ def _ensure_single_series(df: dd.DataFrame, name: str): else: logger.warning(f"{product_name} YAML says instrument_type<-instrument_type_homogenized, but missing.") + if norm(cols_cfg.get("object_type")) == "object_type_homogenized": + if "object_type" in df.columns: + df["object_type_homogenized"] = _ensure_single_series(df, "object_type") + logger.info( + f"{product_name} Using user-provided homogenized object_type via YAML mapping." + ) + else: + logger.warning( + f"{product_name} YAML says object_type<-object_type_homogenized, but missing." + ) + return df @@ -205,12 +444,13 @@ def _homogenize( product_name: Catalog identifier. logger: Logger. type_cast_ok: Whether `type` was normalized. - require_z_flag_homogenized: Create and validate the flag for semantic - star classification even when it is not a ranking priority. + require_z_flag_homogenized: Create and validate the quality flag even + when it is not a ranking priority. Returns: Tuple: (df, used_type_fastpath, tiebreaking_priority, instrument_type_priority, translation_rules_uc) """ + validate_translation_config(translation_config) tiebreaking_priority = translation_config.get("tiebreaking_priority", []) instrument_type_priority = translation_config.get("instrument_type_priority", {}) translation_rules_uc = {k.upper(): v for k, v in translation_config.get("translation_rules", {}).items()} @@ -218,13 +458,63 @@ def _homogenize( z_flag_is_priority = "z_flag_homogenized" in tiebreaking_priority needs_z_flag = z_flag_is_priority or require_z_flag_homogenized + def _fast_path_policy(key: str) -> str: + if "survey" not in df.columns: + return "auto" + surveys = ( + df["survey"].dropna().astype(str).str.upper().unique().compute().tolist() + ) + policies = { + str( + translation_rules_uc.get(survey, {}) + .get(f"{key}_translation", {}) + .get("fast_path", "auto") + ) + for survey in surveys + } + if "disabled" in policies and "required" in policies: + raise ValueError( + f"[{product_name}] Conflicting {key} fast_path policies for surveys " + f"{sorted(surveys)}: cannot mix 'disabled' and 'required' in one catalog" + ) + if "disabled" in policies: + return "disabled" + if "required" in policies: + return "required" + return "auto" + + def _validate_result_domain( + column: str, allowed: set, *, normalize_string: bool = False + ) -> None: + values = df[column] + if normalize_string: + values = values.map_partitions( + _normalize_string_series_to_na, + meta=pd.Series(pd.array([], dtype=DTYPE_STR)), + ).str.lower() + else: + values = dd.to_numeric(values, errors="coerce") + invalid = (~dd.isna(values)) & ~values.isin(list(allowed)) + invalid_count, non_null_count = dask.compute(invalid.sum(), values.count()) + validated_non_null_counts[column] = int(non_null_count) + if int(invalid_count): + examples = df[column].loc[invalid].head(5, compute=True).tolist() + raise ValueError( + f"[{product_name}] Translation produced invalid values in '{column}'. " + f"Allowed set is {sorted(allowed)} (NaN allowed). Examples: {examples}" + ) + # ----------------------- # Vectorized translator # ----------------------- def _translate_column_vectorized(df: dd.DataFrame, key: str, out_col: str, out_kind: str) -> dd.DataFrame: """Apply YAML translation rules per partition.""" - assert key in {"z_flag", "instrument_type"} - assert out_col in {"z_flag_homogenized", "instrument_type_homogenized"} + assert key in {"z_flag", "instrument_type", "object_type"} + assert out_col in { + "z_flag_homogenized", + "instrument_type_homogenized", + "object_type_homogenized", + } assert out_kind in {"float", "str"} def _partition(p: pd.DataFrame) -> pd.DataFrame: @@ -320,35 +610,58 @@ def visit_Compare(self, node): fill_vals = pd.Series([default_val] * int(mask_s.sum()), index=out.index[mask_s], dtype=DTYPE_STR) out.loc[mask_s] = out.loc[mask_s].fillna(fill_vals) - direct = {k: v for k, v in rule.items() if k not in {"conditions", "default"}} + source_col = str(rule.get("source", key)) + optional_source = bool(rule.get("optional_source", False)) + allow_condition_overlap = bool( + rule.get("allow_condition_overlap", False) + ) + direct = { + k: v + for k, v in rule.items() + if k not in { + "conditions", + "default", + "source", + "optional_source", + "allow_condition_overlap", + "fast_path", + } + } if direct: - col = s.loc[mask_s, key] - is_num = pd.api.types.is_numeric_dtype(col) - if key == "z_flag": - is_num = True - if is_num: - col_num = pd.to_numeric(col, errors="coerce") - num_map = {} - for rk, rv in direct.items(): - try: - num_map[float(rk)] = rv - except Exception: - pass - mapped = col_num.map(num_map) + if source_col not in s.columns: + if not optional_source: + raise ValueError( + f"Missing source column '{source_col}' for survey " + f"'{sname}' and translation '{out_col}'." + ) else: - col_str = col.astype(str).str.strip().str.lower() - str_map = {str(k).strip().lower(): v for k, v in direct.items()} - mapped = col_str.map(str_map) - - if out_kind == "float": - out.loc[mask_s] = mapped.fillna(out.loc[mask_s]).astype(DTYPE_FLOAT) - else: - mapped = mapped.astype("object") - mapped_str = pd.Series(pd.array(mapped.where(mapped.notna(), None), dtype=DTYPE_STR), index=mapped.index) - take = mapped_str.notna() - out.loc[take.index] = out.loc[take.index].where(~take, mapped_str) - - for cond in (rule.get("conditions") or []): + col = s.loc[mask_s, source_col] + is_num = pd.api.types.is_numeric_dtype(col) + if key == "z_flag": + is_num = True + if is_num: + col_num = pd.to_numeric(col, errors="coerce") + num_map = {float(rk): rv for rk, rv in direct.items()} + mapped = col_num.map(num_map) + else: + col_str = col.astype(str).str.strip().str.lower() + str_map = {str(k).strip().lower(): v for k, v in direct.items()} + mapped = col_str.map(str_map) + + if out_kind == "float": + out.loc[mask_s] = mapped.fillna(out.loc[mask_s]).astype(DTYPE_FLOAT) + else: + mapped = mapped.astype("object") + mapped_str = pd.Series(pd.array(mapped.where(mapped.notna(), None), dtype=DTYPE_STR), index=mapped.index) + take = mapped_str.notna() + out.loc[take.index] = out.loc[take.index].where(~take, mapped_str) + + condition_values = pd.Series( + pd.array([pd.NA] * len(s), dtype="object"), index=s.index + ) + for condition_index, cond in enumerate( + rule.get("conditions") or [] + ): expr = cond.get("expr") if not expr: continue @@ -389,6 +702,26 @@ def visit_Compare(self, node): mlocal = mlocal & mask_s_aligned val = cond.get("value", default_val) + normalized_val = pd.NA if pd.isna(val) else val + conflicting = ( + mlocal + & condition_values.notna() + & condition_values.ne(normalized_val).fillna(False) + ) + conflict_count = int(conflicting.sum()) + if conflict_count and not allow_condition_overlap: + logger.warning( + "[%s] %s.%s condition %d overwrites a different " + "condition value for %d row(s); later conditions " + "take precedence. Set allow_condition_overlap: true " + "when this is intentional.", + product_name, + sname, + f"{key}_translation", + condition_index + 1, + conflict_count, + ) + condition_values.loc[mlocal] = normalized_val if out_kind == "float": out.loc[mlocal] = pd.to_numeric(val, errors="coerce") else: @@ -444,8 +777,20 @@ def quality_like_to_flag(x): if needs_z_flag: if "z_flag_homogenized" not in df.columns: - if can_use_zflag_as_quality(): - logger.info(f"{product_name} Using 'z_flag' fast path for z_flag_homogenized.") + z_fast_path = _fast_path_policy("z_flag") + z_fast_path_compatible = can_use_zflag_as_quality() + if z_fast_path == "required" and not z_fast_path_compatible: + raise ValueError( + f"[{product_name}] z_flag fast_path is 'required', but z_flag " + "is not a non-empty probability-like column in [0, 1]" + ) + if z_fast_path != "disabled" and z_fast_path_compatible: + logger.info( + "%s Using 'z_flag' fast path for z_flag_homogenized " + "(policy=%s); YAML z_flag rules are bypassed.", + product_name, + z_fast_path, + ) df["z_flag_homogenized"] = df["z_flag"].map_partitions( lambda s: s.apply(quality_like_to_flag).astype(DTYPE_FLOAT), meta=pd.Series(pd.array([], dtype=DTYPE_FLOAT)), @@ -463,14 +808,20 @@ def quality_like_to_flag(x): df = _translate_column_vectorized(df, key="z_flag", out_col="z_flag_homogenized", out_kind="float") + _validate_result_domain( + "z_flag_homogenized", {0.0, 1.0, 2.0, 3.0, 4.0} + ) else: # User-provided 'z_flag_homogenized' is present. Validate allowed domain {0,1,2,3,4} (NaN allowed). logger.info(f"{product_name} 'z_flag_homogenized' already exists; validating user-provided values.") - allowed = {0.0, 1.0, 2.0, 3.0, 4.0, 6.0} + allowed = {0.0, 1.0, 2.0, 3.0, 4.0} vals = dd.to_numeric(df["z_flag_homogenized"], errors="coerce") # NaN is allowed; only non-NaN values outside the allowed set are invalid - invalid_mask = (~dd.isna(vals)) & ~vals.isin(list(allowed)) + invalid_mask = ( + ((~dd.isna(df["z_flag_homogenized"])) & dd.isna(vals)) + | ((~dd.isna(vals)) & ~vals.isin(list(allowed))) + ) invalid_count, non_null_count = dask.compute( invalid_mask.sum(), vals.count() ) @@ -509,8 +860,20 @@ def can_use_type_for_instrument() -> bool: if "instrument_type_homogenized" in tiebreaking_priority: if "instrument_type_homogenized" not in df.columns: - if can_use_type_for_instrument(): - logger.info(f"{product_name} Using 'type' fast path for instrument_type_homogenized.") + instrument_fast_path = _fast_path_policy("instrument_type") + instrument_fast_path_compatible = can_use_type_for_instrument() + if instrument_fast_path == "required" and not instrument_fast_path_compatible: + raise ValueError( + f"[{product_name}] instrument_type fast_path is 'required', " + "but type is not a non-empty column containing only s/g/p" + ) + if instrument_fast_path != "disabled" and instrument_fast_path_compatible: + logger.info( + "%s Using 'type' fast path for instrument_type_homogenized " + "(policy=%s); YAML instrument rules are bypassed.", + product_name, + instrument_fast_path, + ) df["instrument_type_homogenized"] = df["type"].map_partitions( _normalize_string_series_to_na, meta=pd.Series(pd.array([], dtype=DTYPE_STR)), @@ -533,6 +896,11 @@ def can_use_type_for_instrument() -> bool: _normalize_string_series_to_na, meta=pd.Series(pd.array([], dtype=DTYPE_STR)), ).str.lower() + _validate_result_domain( + "instrument_type_homogenized", + {"s", "g", "p"}, + normalize_string=True, + ) else: # User-provided 'instrument_type_homogenized' is present. Validate allowed domain {"s","p","g"}. logger.info(f"{product_name} 'instrument_type_homogenized' already exists; validating user-provided values.") @@ -561,23 +929,48 @@ def can_use_type_for_instrument() -> bool: # Keep normalized lower-case values for consistency df["instrument_type_homogenized"] = normed + # object_type_homogenized is an output-schema field, not a ranking field. + # An entirely-null result is valid for catalogs without classification data. + if "object_type_homogenized" not in df.columns: + df = _translate_column_vectorized( + df, + key="object_type", + out_col="object_type_homogenized", + out_kind="str", + ) + + object_types = df["object_type_homogenized"].map_partitions( + _normalize_string_series_to_na, + meta=pd.Series(pd.array([], dtype=DTYPE_STR)), + ).str.lower() + allowed_object_types = {"star", "galactic", "qso", "agn", "galaxy"} + invalid_object_mask = (~dd.isna(object_types)) & ~object_types.isin( + list(allowed_object_types) + ) + invalid_object_count = int(invalid_object_mask.sum().compute()) + if invalid_object_count: + examples = ( + df["object_type_homogenized"] + .loc[invalid_object_mask] + .head(5, compute=True) + .tolist() + ) + raise ValueError( + f"[{product_name}] Invalid values in 'object_type_homogenized'. " + f"Allowed set is {sorted(allowed_object_types)} (NaN allowed). " + f"Examples of invalid values: {examples}" + ) + df["object_type_homogenized"] = object_types + # --- post-homogenization sanity checks (required columns must not be all-NaN) --- if needs_z_flag: if "z_flag_homogenized" not in df.columns: raise ValueError( f"[{product_name}] 'z_flag_homogenized' is required for " - "star classification but is missing after homogenization." + "quality ranking or filtering but is missing after homogenization." ) - if z_flag_is_priority: - non_null = validated_non_null_counts.get("z_flag_homogenized") - if non_null is None: - non_null = dask.compute(df["z_flag_homogenized"].count())[0] - if int(non_null) == 0: - raise ValueError( - f"[{product_name}] All values in 'z_flag_homogenized' are NaN. " - "This column is required (in tiebreaking_priority) and must contain at least one non-NaN value. " - "Verify YAML translations / fast-path logic and input columns." - ) + # An all-null quality column is valid: this priority simply cannot + # distinguish candidates. Later priorities or hard-tie handling apply. if "instrument_type_homogenized" in tiebreaking_priority: if "instrument_type_homogenized" not in df.columns: diff --git a/scripts/crd-run.py b/scripts/crd-run.py index 7a58c45..17ed891 100644 --- a/scripts/crd-run.py +++ b/scripts/crd-run.py @@ -30,7 +30,6 @@ import dask import dask.dataframe as dd import lsdb -import numpy as np import pandas as pd # ----------------------- @@ -42,17 +41,24 @@ from dask.distributed import wait as dask_wait from deduplication import ( REPRESENTATIVE_RADIUS_DIAGNOSTIC_COLUMN, + build_global_tie_invariant_diagnostics, count_global_edge_group_mismatches, count_global_tie_invariant_violations, filter_dask_by_tie_treatment, filter_pandas_by_tie_treatment, run_dedup_with_lsdb_map_partitions, + validate_object_type_inclusion, validate_spatial_safety, ) from executor import get_executor from product_handle import save_dataframe from resource_usage import ResourceUsageMonitor -from specz import prepare_catalog, validate_combine_configuration +from specz import ( + build_runtime_schema_hints, + prepare_catalog, + validate_combine_configuration, +) +from specz_homogenization import validate_translation_config from utils import ( configure_exception_hook, configure_warning_handler, @@ -725,6 +731,7 @@ def main( try: translation_config = load_yml(path_to_translation_file) + validate_translation_config(translation_config) except Exception as e: log_init.error("Failed to parse flags_translation_file: %s", e, exc_info=True) raise @@ -737,15 +744,28 @@ def main( margin_warning_fraction = float( translation_config.get("margin_warning_fraction", 0.8) ) + configured_max_radius = translation_config.get( + "max_representative_radius_arcsec", None + ) + max_representative_radius_arcsec = ( + None if configured_max_radius is None else float(configured_max_radius) + ) validate_spatial_safety( crossmatch_radius_arcsec, margin_threshold_arcsec, margin_warning_fraction, + max_representative_radius_arcsec, ) log_init.info( - 'Spatial safety: crossmatch_radius=%.3f" margin_threshold=%.3f" ratio=%.3f', + 'Spatial safety: crossmatch_radius=%.3f" margin_threshold=%.3f" ' + "max_representative_radius=%s ratio=%.3f", crossmatch_radius_arcsec, margin_threshold_arcsec, + ( + "disabled" + if max_representative_radius_arcsec is None + else f'{max_representative_radius_arcsec:.3f}"' + ), crossmatch_radius_arcsec / margin_threshold_arcsec, ) # Minimal summary of loaded translation (no heavy dumping) @@ -780,6 +800,36 @@ def main( log_init, ) translation_config["tiebreaking_priority"] = validated_priorities + object_type_inclusion = validate_object_type_inclusion( + { + key: param_config[key] + for key in ( + "include_unclassified", + "include_galaxy", + "include_star", + "include_agn", + "include_qso", + "include_galactic", + ) + if key in param_config + } + ) + log_init.info("Object-type graph inclusion: %s", object_type_inclusion) + if ( + cut_value_numeric in {1.0, 2.0, 3.0, 4.0} + and ( + object_type_inclusion["include_star"] + or object_type_inclusion["include_galactic"] + ) + ): + log_init.warning( + "An active z_flag_homogenized cut removes star/galactic rows whose " + "quality is null before graph inclusion. Their internal 0.5 ranking " + "fallback applies only to rows that survive this cut." + ) + translation_config["runtime_schema_hints"] = build_runtime_schema_hints( + param_config, translation_config + ) completed = read_completed_steps(os.path.join(temp_dir, "process_resume.log")) # --- Dask cluster/client --- @@ -1634,7 +1684,7 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: ####################################################################### # Diagnostics / outputs - # - edge_log: enable edge diagnostics (warn on star-neighbor exclusions) + # - edge_log: enable diagnostics for object types excluded from the graph # - group_col: set to None to disable exporting group labels edge_log = bool( translation_config.get("dedup_edge_diagnostics_enabled", False) @@ -1691,6 +1741,8 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: margin_threshold_arcsec=margin_threshold_arcsec, margin_warning_fraction=margin_warning_fraction, representative_radius_diagnostics_enabled=representative_radius_diagnostics_enabled, + object_type_inclusion=object_type_inclusion, + max_representative_radius_arcsec=max_representative_radius_arcsec, ) log_dedup.info( "Labels graph built (lazy). Persisting compact labels for " @@ -1779,6 +1831,16 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: # Coalesce tie_result in Dask (still lazy) try: + label_merge_diagnostics = bool( + translation_config.get( + "label_merge_diagnostics_enabled", True + ) + ) + missing_new_labels_lazy = ( + merged["tie_result_new"].isna().sum() + if label_merge_diagnostics + else None + ) if "tie_result" in merged.columns: merged["tie_result"] = merged["tie_result_new"].fillna( merged["tie_result"] @@ -1787,15 +1849,7 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: merged = merged.assign(tie_result=merged["tie_result_new"]) if "tie_result_new" in merged.columns: merged = merged.drop(columns=["tie_result_new"]) - if "z_flag_homogenized" in merged.columns: - star_mask = dd.to_numeric( - merged["z_flag_homogenized"], errors="coerce" - ).eq(6.0) - merged["tie_result"] = ( - merged["tie_result"] - .mask(star_mask, np.int8(3)) - .astype("Int8") - ) + merged["tie_result"] = merged["tie_result"].astype("Int8") validate_edges = bool( translation_config.get("validate_global_graph_edges", False) @@ -1803,7 +1857,23 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: validate_ties = bool( translation_config.get("validate_global_tie_invariants", False) ) + tie_diagnostics = bool( + translation_config.get( + "tie_invariant_diagnostics_enabled", True + ) + ) + detailed_tie_diagnostics = bool( + translation_config.get( + "tie_invariant_diagnostics_detailed_enabled", False + ) + ) + invalid_stats_lazy = None validation_tasks = [] + representative_limit_arcsec = ( + max_representative_radius_arcsec + if max_representative_radius_arcsec is not None + else crossmatch_radius_arcsec + ) if representative_radius_diagnostics_enabled: representative_radius = labels_dd[ REPRESENTATIVE_RADIUS_DIAGNOSTIC_COLUMN @@ -1813,10 +1883,10 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: [ representative_radius_valid.count(), representative_radius_valid.gt( - crossmatch_radius_arcsec + representative_limit_arcsec ).sum(), representative_radius_valid.gt( - 2.0 * crossmatch_radius_arcsec + 2.0 * representative_limit_arcsec ).sum(), representative_radius_valid.gt( margin_threshold_arcsec @@ -1829,13 +1899,31 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: count_global_edge_group_mismatches(merged) ) validation_tasks.extend([mismatch_lazy, dangling_lazy]) + if label_merge_diagnostics: + validation_tasks.append(missing_new_labels_lazy) if validate_ties: - invalid_groups_lazy = count_global_tie_invariant_violations( - labels_dd, - z_flag_col=None, - ) - - validation_tasks.append(invalid_groups_lazy) + if tie_diagnostics or detailed_tie_diagnostics: + invalid_stats_lazy, missing_group_rows_lazy = ( + build_global_tie_invariant_diagnostics(labels_dd) + ) + validation_tasks.extend( + [ + invalid_stats_lazy.map_partitions(len).sum() + + missing_group_rows_lazy, + missing_group_rows_lazy, + invalid_stats_lazy["multiple_winners"].sum(), + invalid_stats_lazy["no_survivor"].sum(), + invalid_stats_lazy[ + "mixed_winner_hard_tie" + ].sum(), + invalid_stats_lazy["single_hard_tie"].sum(), + invalid_stats_lazy["invalid_tie_values"].sum(), + ] + ) + else: + validation_tasks.append( + count_global_tie_invariant_violations(labels_dd) + ) validation_results = iter(dask.compute(*validation_tasks)) if representative_radius_diagnostics_enabled: @@ -1857,7 +1945,7 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: "fraction_exceeding=%.6f exceeding_twice_radius=%d " "exceeding_margin=%d max_radius=%.4farcsec", representative_components, - crossmatch_radius_arcsec, + representative_limit_arcsec, representative_exceed_radius, representative_fraction, representative_exceed_twice_radius, @@ -1869,16 +1957,23 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: dangling_count = int(next(validation_results)) log_dedup.info( "Global graph validation: cross_group_edges=%d " - "dangling_nonstar_edges=%d", + "dangling_participating_edges=%d", mismatch_count, dangling_count, ) if mismatch_count: - raise RuntimeError( - "Partition-local deduplication produced " - f"{mismatch_count} non-star edges whose endpoints " - "have different canonical group_id values. Increase " - "margin_threshold_arcsec or inspect long components." + if max_representative_radius_arcsec is None: + raise RuntimeError( + "Partition-local deduplication produced " + f"{mismatch_count} participating edges whose " + "endpoints have different canonical group_id " + "values. Increase margin_threshold_arcsec or " + "inspect long components." + ) + log_dedup.info( + "Cross-group edges are expected with representative-" + "radius truncation enabled (radius=%.3f arcsec).", + max_representative_radius_arcsec, ) if dangling_count: raise RuntimeError( @@ -1889,12 +1984,81 @@ def _submit_pair(executor: ThreadPoolExecutor) -> bool: "catalog concatenation." ) + if label_merge_diagnostics: + missing_new_labels = int(next(validation_results)) + log_dedup.info( + "Label merge diagnostics: rows_without_new_label=%d", + missing_new_labels, + ) + if validate_ties: invalid_groups = int(next(validation_results)) + if tie_diagnostics or detailed_tie_diagnostics: + missing_group_rows = int(next(validation_results)) + multiple_winners = int(next(validation_results)) + no_survivor = int(next(validation_results)) + mixed_winner_hard_tie = int(next(validation_results)) + single_hard_tie = int(next(validation_results)) + invalid_tie_values = int(next(validation_results)) + log_dedup.info( + "Global tie invariant diagnostics: " + "missing_group_rows=%d multiple_winners=%d " + "no_survivor=%d mixed_winner_hard_tie=%d " + "single_hard_tie=%d invalid_tie_values=%d", + missing_group_rows, + multiple_winners, + no_survivor, + mixed_winner_hard_tie, + single_hard_tie, + invalid_tie_values, + ) log_dedup.info( "Global tie invariant validation: invalid_groups=%d", invalid_groups, ) + if ( + invalid_groups + and detailed_tie_diagnostics + and invalid_stats_lazy is not None + ): + sample_size = int( + translation_config.get( + "tie_invariant_diagnostics_sample_size", 10 + ) + ) + max_rows = int( + translation_config.get( + "tie_invariant_diagnostics_max_rows", 100 + ) + ) + invalid_sample = invalid_stats_lazy.reset_index().head( + sample_size, npartitions=-1 + ) + log_dedup.error( + "Invalid tie group summary sample: %s", + invalid_sample.to_dict("records"), + ) + sample_group_ids = invalid_sample["group_id"].tolist() + if sample_group_ids: + member_columns = [ + column + for column in ( + "group_id", + "CRD_ID", + "tie_result", + "z_flag_homogenized", + "object_type_homogenized", + ) + if column in labels_dd.columns + ] + member_sample = labels_dd.loc[ + labels_dd["group_id"].isin(sample_group_ids), + member_columns, + ].head(max_rows, npartitions=-1) + log_dedup.error( + "Invalid tie group member sample: %s", + member_sample.to_dict("records"), + ) if invalid_groups: raise RuntimeError( f"Global tie-result invariants failed for " diff --git a/tests/test_column_mapping.py b/tests/test_column_mapping.py index 24689c2..2f67997 100644 --- a/tests/test_column_mapping.py +++ b/tests/test_column_mapping.py @@ -72,6 +72,7 @@ def test_column_mapping_ignores_null_mapping_and_creates_base_schema(): "z", "z_flag", "z_err", + "object_type", "source", } assert expected.issubset(result.columns) diff --git a/tests/test_crossmatch_dtype_normalization.py b/tests/test_crossmatch_dtype_normalization.py new file mode 100644 index 0000000..7186d81 --- /dev/null +++ b/tests/test_crossmatch_dtype_normalization.py @@ -0,0 +1,90 @@ +"""Regression tests for strict crossmatch schema normalization.""" + +import sys +import types +from pathlib import Path + +import pandas as pd + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "packages")) +if "tables_io" not in sys.modules: + tables_io = types.ModuleType("tables_io") + tables_io.types = types.SimpleNamespace(PD_DATAFRAME="PD_DATAFRAME") + sys.modules["tables_io"] = tables_io + +from crossmatch_cross import ( # noqa: E402 + _EXPECTED_TYPES, + _cast_partition_expected, + _get_expr_schema_hints, +) +from specz import DTYPE_STR, build_runtime_schema_hints # noqa: E402 + + +def test_object_type_string_backends_normalize_to_same_dtype(): + python_string = pd.DataFrame( + {"object_type_homogenized": pd.Series(["star"], dtype="string")} + ) + pyarrow_string = pd.DataFrame( + {"object_type_homogenized": pd.Series(["agn"], dtype="string[pyarrow]")} + ) + + left = _cast_partition_expected(python_string, _EXPECTED_TYPES, None) + right = _cast_partition_expected(pyarrow_string, _EXPECTED_TYPES, None) + + assert left["object_type_homogenized"].dtype == DTYPE_STR + assert right["object_type_homogenized"].dtype == DTYPE_STR + assert left["object_type_homogenized"].dtype == right["object_type_homogenized"].dtype + + +def test_runtime_schema_combines_expr_extra_priority_and_enabled_flags(): + param_config = { + "extra_columns": { + "label": "str", + "external_score": {"source": "RAW_SCORE", "type": "float"}, + }, + "insert_DP1_footprint_flag": True, + "insert_rubin_footprint_flag": False, + } + translation_config = { + "save_expr_columns": True, + "expr_column_schema": {"expr_value": "int", "label": "float"}, + "tiebreaking_priority": ["z_flag_homogenized", "custom_rank"], + } + + hints = build_runtime_schema_hints(param_config, translation_config) + + assert hints == { + "expr_value": "int", + "label": "str", + "external_score": "float", + "is_in_DP1_fields": "int", + "custom_rank": "float", + } + + +def test_crossmatch_runtime_hints_override_expr_hints_and_normalize_extra_columns(): + config = { + "save_expr_columns": True, + "expr_column_schema": {"label": "float", "expr_value": "int"}, + "runtime_schema_hints": {"label": "str", "extra_text": "str"}, + } + hints = _get_expr_schema_hints(config) + assert hints == {"label": "str", "expr_value": "int", "extra_text": "str"} + + left = pd.DataFrame( + { + "label": pd.Series(["left"], dtype="string"), + "extra_text": pd.Series(["a"], dtype="string"), + } + ) + right = pd.DataFrame( + { + "label": pd.Series(["right"], dtype="string[pyarrow]"), + "extra_text": pd.Series(["b"], dtype="string[pyarrow]"), + } + ) + left_fixed = _cast_partition_expected(left, _EXPECTED_TYPES, hints) + right_fixed = _cast_partition_expected(right, _EXPECTED_TYPES, hints) + + assert left_fixed["label"].dtype == right_fixed["label"].dtype == DTYPE_STR + assert left_fixed["extra_text"].dtype == right_fixed["extra_text"].dtype == DTYPE_STR diff --git a/tests/test_dedup_partition_safety.py b/tests/test_dedup_partition_safety.py index 0a642f5..9083267 100644 --- a/tests/test_dedup_partition_safety.py +++ b/tests/test_dedup_partition_safety.py @@ -10,16 +10,18 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "packages")) from deduplication import ( # noqa: E402 + SPLIT_REFERENCE_COLUMN, _collapse_within_dz, _dedup_local_with_margin, _log_representative_radius_diagnostics, _validate_local_tie_invariants, + build_global_tie_invariant_diagnostics, count_global_edge_group_mismatches, count_global_tie_invariant_violations, ) -def _row(crd_id, neighbor, ra, flag): +def _row(crd_id, neighbor, ra, flag, object_type=pd.NA): return { "CRD_ID": crd_id, "compared_to": neighbor, @@ -28,6 +30,7 @@ def _row(crd_id, neighbor, ra, flag): "z": 0.1, "z_flag_homogenized": flag, "tie_result": 1, + "object_type_homogenized": object_type, } @@ -55,7 +58,38 @@ def test_boundary_component_has_same_canonical_group_from_both_pixels(): assert from_b_pixel["tie_result"] == 0 -def test_custom_priority_keeps_star_semantics_without_flag_ranking(): +def test_reference_radius_split_is_stable_across_main_margin_views(): + rows = [ + _row("A", "B", 10.0, 4.0), + _row("B", "A, C", 10.0 + 0.5 / 3600.0, 3.0), + _row("C", "B, D", 10.0 + 1.0 / 3600.0, 2.0), + _row("D", "C", 10.0 + 1.5 / 3600.0, 1.0), + ] + first = _dedup_local_with_margin( + pd.DataFrame(rows[:2]), + pd.DataFrame(rows[2:]), + pixel=None, + tiebreaking_priority=["z_flag_homogenized"], + instrument_type_priority=None, + group_col="group_id", + max_representative_radius_arcsec=1.0, + ).set_index("CRD_ID") + second = _dedup_local_with_margin( + pd.DataFrame(rows[2:]), + pd.DataFrame(rows[:2]), + pixel=None, + tiebreaking_priority=["z_flag_homogenized"], + instrument_type_priority=None, + group_col="group_id", + max_representative_radius_arcsec=1.0, + ).set_index("CRD_ID") + + assert first.loc["A", "group_id"] == first.loc["B", "group_id"] + assert second.loc["C", "group_id"] == first.loc["A", "group_id"] + assert second.loc["D", "group_id"] != first.loc["A", "group_id"] + + +def test_custom_priority_keeps_excluded_star_outside_graph(): rows = [ { **_row("A", "B, S", 10.0, 4.0), @@ -66,7 +100,7 @@ def test_custom_priority_keeps_star_semantics_without_flag_ranking(): "custom_score": 10.0, }, { - **_row("S", "A, B", 10.0 + 0.2 / 3600.0, 6.0), + **_row("S", "A, B", 10.0 + 0.2 / 3600.0, None, "star"), "custom_score": 100.0, }, ] @@ -89,7 +123,7 @@ def test_local_invariant_accepts_single_winner_and_hard_tie(): frame = pd.DataFrame( { "group_id": [1, 1, 2, 2, 3], - "z_flag_homogenized": [4, 3, 4, 4, 6], + "z_flag_homogenized": [4, 3, 4, 4, pd.NA], "tie_result": [1, 0, 2, 2, 3], } ) @@ -170,7 +204,7 @@ def test_global_tie_validation_detects_invalid_patterns(): pd.DataFrame( { "group_id": [1, 1, 2, 2, 3, 3, 4], - "z_flag_homogenized": [4, 3, 4, 4, 4, 4, 6], + "z_flag_homogenized": [4, 3, 4, 4, 4, 4, pd.NA], "tie_result": [1, 0, 2, 2, 1, 1, 3], } ), @@ -180,6 +214,12 @@ def test_global_tie_validation_detects_invalid_patterns(): assert count_global_tie_invariant_violations(frame).compute() == 1 + diagnostics, missing_group_rows = build_global_tie_invariant_diagnostics(frame) + computed = diagnostics.compute() + assert missing_group_rows.compute() == 0 + assert len(computed) == 1 + assert bool(computed.iloc[0]["multiple_winners"]) + def test_global_tie_validation_accepts_compact_labels_without_z_flag(): labels = dd.from_pandas( @@ -228,3 +268,31 @@ def test_representative_radius_diagnostic_warns_for_transitive_chain(): assert "Representative-radius diagnostics" in logger.warning.call_args.args[0] assert diagnostic.notna().sum() == 1 assert diagnostic.max() == pytest.approx(0.8, abs=1e-6) + + +def test_representative_radius_diagnostic_uses_split_reference_not_winner(): + frame = pd.DataFrame( + { + "group_id": [1, 1, 1], + "tie_result": [0, 1, 0], + "CRD_ID": ["reference", "winner", "edge"], + SPLIT_REFERENCE_COLUMN: ["reference"] * 3, + "ra": [10.0, 10.0 + 0.8 / 3600.0, 10.0 - 0.8 / 3600.0], + "dec": [0.0, 0.0, 0.0], + } + ) + logger = Mock() + + with patch("deduplication._phase_logger", return_value=logger): + diagnostic = _log_representative_radius_diagnostics( + frame, + group_col="group_id", + tie_col="tie_result", + crd_col="CRD_ID", + radius_arcsec=1.0, + partition_tag="test", + ) + + logger.warning.assert_not_called() + assert diagnostic.notna().sum() == 1 + assert diagnostic.max() == pytest.approx(0.8, abs=1e-6) diff --git a/tests/test_deduplication_core.py b/tests/test_deduplication_core.py index 8a82bdb..0442d2f 100644 --- a/tests/test_deduplication_core.py +++ b/tests/test_deduplication_core.py @@ -6,18 +6,30 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "packages")) -from deduplication import deduplicate_pandas # noqa: E402 +from deduplication import ( # noqa: E402 + deduplicate_pandas, + validate_object_type_inclusion, +) INSTRUMENT_PRIORITY = {"s": 3, "g": 2, "p": 1} -def _row(crd_id, compared_to, *, flag=4.0, instrument="s", z=0.1): +def _row( + crd_id, + compared_to, + *, + flag=4.0, + instrument="s", + z=0.1, + object_type=pd.NA, +): return { "CRD_ID": crd_id, "compared_to": compared_to, "z": z, "z_flag_homogenized": flag, "instrument_type_homogenized": instrument, + "object_type_homogenized": object_type, } @@ -68,11 +80,11 @@ def test_full_dedup_emits_two_and_three_way_hard_ties(): assert three_way["group_id"].nunique() == 1 -def test_stars_are_isolated_from_nonstar_deduplication(): +def test_excluded_stars_are_isolated_from_deduplication(): result = _deduplicate( [ _row("A", "S", flag=4), - _row("S", "A, B", flag=6), + _row("S", "A, B", flag=None, object_type="star"), _row("B", "S", flag=3), ] ) @@ -132,8 +144,8 @@ def test_singleton_and_dangling_neighbor_remain_winners(): def test_catalog_containing_only_stars_keeps_each_star_isolated(): result = _deduplicate( [ - _row("S1", "S2", flag=6), - _row("S2", "S1", flag=6), + _row("S1", "S2", flag=None, object_type="star"), + _row("S2", "S1", flag=None, object_type="star"), ] ) @@ -141,6 +153,152 @@ def test_catalog_containing_only_stars_keeps_each_star_isolated(): assert result["group_id"].nunique() == 2 +def test_included_star_without_quality_uses_internal_half_point_score(): + result = deduplicate_pandas( + pd.DataFrame( + [ + _row("S", "G", flag=None, object_type="star"), + _row("G", "S", flag=0, object_type="galaxy"), + ] + ), + tiebreaking_priority=["z_flag_homogenized"], + object_type_inclusion={"include_star": True}, + delta_z_threshold=0, + ).set_index("CRD_ID") + + assert result["tie_result"].astype(int).to_dict() == {"S": 1, "G": 0} + assert pd.isna(result.loc["S", "z_flag_homogenized"]) + + +def test_internal_half_point_loses_to_real_quality_one(): + result = deduplicate_pandas( + pd.DataFrame( + [ + _row("S", "G", flag=None, object_type="star"), + _row("G", "S", flag=1, object_type="galaxy"), + ] + ), + tiebreaking_priority=["z_flag_homogenized"], + object_type_inclusion={"include_star": True}, + delta_z_threshold=0, + ).set_index("CRD_ID") + + assert result["tie_result"].astype(int).to_dict() == {"S": 0, "G": 1} + + +def test_object_type_inclusion_requires_booleans_and_one_enabled_type(): + with pytest.raises(TypeError, match=r"param\.include_star must be a boolean"): + validate_object_type_inclusion({"include_star": "yes"}) + with pytest.raises(ValueError, match="At least one"): + validate_object_type_inclusion( + { + "include_unclassified": False, + "include_galaxy": False, + "include_star": False, + "include_agn": False, + "include_qso": False, + "include_galactic": False, + } + ) + + +@pytest.mark.parametrize("object_type", ["star", "galactic"]) +def test_default_object_type_policy_excludes_stellar_and_galactic(object_type): + result = _deduplicate( + [ + _row("X", "G", flag=None, object_type=object_type), + _row("G", "X", flag=4, object_type="galaxy"), + ] + ) + + assert result.loc["X", "tie_result"] == 3 + assert result.loc["G", "tie_result"] == 1 + assert result["group_id"].nunique() == 2 + + +def test_unclassified_rows_can_be_excluded_explicitly(): + result = deduplicate_pandas( + pd.DataFrame( + [ + _row("U", "G", flag=4, object_type=pd.NA), + _row("G", "U", flag=4, object_type="galaxy"), + ] + ), + tiebreaking_priority=["z_flag_homogenized"], + object_type_inclusion={"include_unclassified": False}, + group_col="group_id", + ).set_index("CRD_ID") + + assert result.loc["U", "tie_result"] == 3 + assert result.loc["G", "tie_result"] == 1 + + +def _chain_rows(): + return [ + { + **_row("A", "B", flag=4, z=0.1), + "ra": 10.0, + "dec": 0.0, + }, + { + **_row("B", "A, C", flag=3, z=0.1), + "ra": 10.0 + 0.5 / 3600.0, + "dec": 0.0, + }, + { + **_row("C", "B, D", flag=2, z=0.1), + "ra": 10.0 + 1.0 / 3600.0, + "dec": 0.0, + }, + { + **_row("D", "C", flag=1, z=0.1), + "ra": 10.0 + 1.5 / 3600.0, + "dec": 0.0, + }, + ] + + +def test_reference_radius_splits_long_transitive_chain(): + result = deduplicate_pandas( + pd.DataFrame(_chain_rows()), + tiebreaking_priority=["z_flag_homogenized"], + max_representative_radius_arcsec=1.0, + delta_z_threshold=0, + group_col="group_id", + ).set_index("CRD_ID") + + assert result.loc[["A", "B", "C"], "group_id"].nunique() == 1 + assert result.loc["D", "group_id"] != result.loc["A", "group_id"] + assert result["tie_result"].astype(int).to_dict() == { + "A": 1, + "B": 0, + "C": 0, + "D": 1, + } + + +def test_reference_radius_split_is_independent_of_row_order(): + forward = deduplicate_pandas( + pd.DataFrame(_chain_rows()), + tiebreaking_priority=["z_flag_homogenized"], + max_representative_radius_arcsec=1.0, + delta_z_threshold=0, + group_col="group_id", + ).set_index("CRD_ID") + reverse = deduplicate_pandas( + pd.DataFrame(list(reversed(_chain_rows()))), + tiebreaking_priority=["z_flag_homogenized"], + max_representative_radius_arcsec=1.0, + delta_z_threshold=0, + group_col="group_id", + ).set_index("CRD_ID") + + assert forward["group_id"].to_dict() == reverse["group_id"].to_dict() + assert forward["tie_result"].astype(int).to_dict() == reverse[ + "tie_result" + ].astype(int).to_dict() + + def test_missing_priority_column_fails_clearly(): with pytest.raises(KeyError, match="unknown_priority"): deduplicate_pandas( @@ -149,7 +307,7 @@ def test_missing_priority_column_fails_clearly(): ) -def test_missing_semantic_star_flag_fails_clearly(): +def test_missing_quality_column_fails_clearly(): frame = pd.DataFrame( { "CRD_ID": ["A"], diff --git a/tests/test_entrypoint_imports.py b/tests/test_entrypoint_imports.py new file mode 100644 index 0000000..95ad749 --- /dev/null +++ b/tests/test_entrypoint_imports.py @@ -0,0 +1,18 @@ +"""Smoke tests for executable entrypoint imports.""" + +import importlib.util +import sys +from pathlib import Path + + +def test_crd_run_module_imports_successfully(): + root = Path(__file__).resolve().parents[1] + packages = str(root / "packages") + if packages not in sys.path: + sys.path.insert(0, packages) + + script = root / "scripts" / "crd-run.py" + spec = importlib.util.spec_from_file_location("crd_run_import_smoke", script) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) diff --git a/tests/test_homogenization.py b/tests/test_homogenization.py index ebe14b1..a70031e 100644 --- a/tests/test_homogenization.py +++ b/tests/test_homogenization.py @@ -7,6 +7,7 @@ import dask.dataframe as dd import pandas as pd import pytest +import yaml sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "packages")) if "tables_io" not in sys.modules: @@ -19,11 +20,103 @@ _requires_z_flag_homogenization, validate_combine_configuration, ) -from specz_homogenization import _homogenize # noqa: E402 +from specz_homogenization import ( + _homogenize, # noqa: E402 + validate_translation_config, # noqa: E402 +) LOGGER = logging.getLogger("test.homogenization") +def test_translation_schema_rejects_unsafe_expression_with_exact_path(): + config = { + "translation_rules": { + "DEMO": { + "object_type_translation": { + "conditions": [ + {"expr": "__import__('os').system('id')", "value": "star"} + ] + } + } + } + } + + with pytest.raises( + ValueError, + match=r"translation_rules\.DEMO\.object_type_translation\.conditions\[0\]\.expr", + ): + validate_translation_config(config) + + +def test_translation_schema_rejects_private_attributes_and_typos(): + private = { + "translation_rules": { + "DEMO": { + "object_type_translation": { + "conditions": [{"expr": "value.__class__", "value": "star"}] + } + } + } + } + with pytest.raises(ValueError, match="private attribute"): + validate_translation_config(private) + + typo = { + "translation_rules": { + "DEMO": { + "object_type_translation": { + "allow_condition_overlaps": True, + "default": None, + } + } + } + } + with pytest.raises(ValueError, match="did you mean 'allow_condition_overlap'"): + validate_translation_config(typo) + + +def test_translation_schema_validates_output_domains_and_condition_shape(): + invalid_value = { + "translation_rules": { + "DEMO": {"z_flag_translation": {"default": 9}} + } + } + with pytest.raises(ValueError, match=r"z_flag_translation\.default=9"): + validate_translation_config(invalid_value) + + invalid_condition = { + "translation_rules": { + "DEMO": { + "object_type_translation": { + "conditions": [{"expr": "value == 1", "vale": "star"}] + } + } + } + } + with pytest.raises(ValueError, match=r"conditions\[0\].*unknown option"): + validate_translation_config(invalid_condition) + + +def test_translation_schema_validates_diagnostic_controls(): + with pytest.raises(TypeError, match="tie_invariant_diagnostics_enabled"): + validate_translation_config({"tie_invariant_diagnostics_enabled": "yes"}) + with pytest.raises(ValueError, match="tie_invariant_diagnostics_sample_size"): + validate_translation_config({"tie_invariant_diagnostics_sample_size": 0}) + with pytest.raises(ValueError, match="max_representative_radius_arcsec"): + validate_translation_config({"max_representative_radius_arcsec": 0}) + + validate_translation_config( + { + "tie_invariant_diagnostics_enabled": True, + "tie_invariant_diagnostics_detailed_enabled": False, + "tie_invariant_diagnostics_sample_size": 10, + "tie_invariant_diagnostics_max_rows": 100, + "label_merge_diagnostics_enabled": True, + "max_representative_radius_arcsec": 1.0, + } + ) + + def test_yaml_flag_translation_applies_direct_default_and_condition(): frame = dd.from_pandas( pd.DataFrame( @@ -54,6 +147,672 @@ def test_yaml_flag_translation_applies_direct_default_and_condition(): assert result.compute()["z_flag_homogenized"].astype(float).tolist() == [2, 0, 4] +def test_z_flag_fast_path_policy_can_be_disabled_or_required(): + frame = dd.from_pandas( + pd.DataFrame({"survey": ["demo", "demo"], "z_flag": [0.2, 0.95]}), + npartitions=1, + sort=False, + ) + disabled = { + "tiebreaking_priority": ["z_flag_homogenized"], + "translation_rules": { + "DEMO": { + "z_flag_translation": {"fast_path": "disabled", "default": 3} + } + }, + } + result, *_ = _homogenize( + frame, disabled, "demo", LOGGER, type_cast_ok=False + ) + assert result.compute()["z_flag_homogenized"].tolist() == [3.0, 3.0] + + incompatible = dd.from_pandas( + pd.DataFrame({"survey": ["demo"], "z_flag": [4]}), + npartitions=1, + sort=False, + ) + required = { + "tiebreaking_priority": ["z_flag_homogenized"], + "translation_rules": { + "DEMO": { + "z_flag_translation": {"fast_path": "required", "default": 3} + } + }, + } + with pytest.raises(ValueError, match="fast_path is 'required'"): + _homogenize(incompatible, required, "demo", LOGGER, type_cast_ok=False) + + +def test_instrument_fast_path_can_be_disabled(): + frame = dd.from_pandas( + pd.DataFrame( + {"survey": ["demo", "demo"], "instrument_type": [pd.NA] * 2, "type": ["s", "g"]} + ), + npartitions=1, + sort=False, + ) + config = { + "tiebreaking_priority": ["instrument_type_homogenized"], + "translation_rules": { + "DEMO": { + "instrument_type_translation": { + "fast_path": "disabled", + "default": "p", + } + } + }, + } + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=True) + assert result.compute()["instrument_type_homogenized"].tolist() == ["p", "p"] + + +def test_optional_source_absence_keeps_condition_fallback(): + frame = dd.from_pandas( + pd.DataFrame( + {"survey": ["demo", "demo"], "object_type": [pd.NA] * 2, "fallback": [1, 0]} + ), + npartitions=1, + sort=False, + ) + config = { + "translation_rules": { + "DEMO": { + "object_type_translation": { + "source": "missing_optional_column", + "optional_source": True, + "STAR": "star", + "conditions": [{"expr": "fallback == 1", "value": "galaxy"}], + "default": None, + } + } + } + } + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "galaxy", + "missing", + ] + + +def test_later_condition_wins_and_conflicting_overlap_warns(caplog): + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["demo"], + "object_type": [pd.NA], + "score": [2], + } + ), + npartitions=1, + sort=False, + ) + config = { + "translation_rules": { + "DEMO": { + "object_type_translation": { + "conditions": [ + {"expr": "score > 0", "value": "star"}, + {"expr": "score > 1", "value": "galaxy"}, + ], + "default": None, + } + } + } + } + + with caplog.at_level(logging.WARNING, logger=LOGGER.name): + result, *_ = _homogenize( + frame, config, "demo", LOGGER, type_cast_ok=False + ) + + assert result.compute()["object_type_homogenized"].tolist() == ["galaxy"] + assert "later conditions take precedence" in caplog.text + + +def test_declared_condition_overlap_does_not_warn(caplog): + frame = dd.from_pandas( + pd.DataFrame( + {"survey": ["demo"], "object_type": [pd.NA], "score": [2]} + ), + npartitions=1, + sort=False, + ) + config = { + "translation_rules": { + "DEMO": { + "object_type_translation": { + "allow_condition_overlap": True, + "conditions": [ + {"expr": "score > 0", "value": "star"}, + {"expr": "score > 1", "value": "galaxy"}, + ], + "default": None, + } + } + } + } + + with caplog.at_level(logging.WARNING, logger=LOGGER.name): + result, *_ = _homogenize( + frame, config, "demo", LOGGER, type_cast_ok=False + ) + + assert result.compute()["object_type_homogenized"].tolist() == ["galaxy"] + assert "later conditions take precedence" not in caplog.text + + +def test_cosmos_web_flag_translation_compares_normalized_string_type(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = ["z_flag_homogenized"] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["COSMOS_Web"] * 5, + "object_type": [pd.NA] * 5, + "type": ["1", "0", "0", "0", "0"], + "z_flag": [0, 1, 0, 0, 0], + "zpdf_med": [1.0, 1.0, 1.01, 1.0, 1.0], + "zchi2": [1.0, 1.0, 1.0, 1.0, 1.0], + "nbfilt": [40, 40, 40, 20, 30], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["z_flag_homogenized"].fillna(-1).tolist() == [ + -1.0, + 0.0, + 1.0, + 2.0, + 3.0, + ] + + +def test_object_type_is_always_present_and_may_be_entirely_null(): + frame = dd.from_pandas( + pd.DataFrame({"survey": ["unknown", "unknown"]}), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, {}, "demo", LOGGER, type_cast_ok=False) + + computed = result.compute() + assert "object_type_homogenized" in computed + assert computed["object_type_homogenized"].isna().all() + + +def test_object_type_translation_uses_canonical_renamed_z_flag(): + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["demo", "demo", "demo"], + "object_type": [pd.NA, pd.NA, pd.NA], + "z_flag": [13, 4, 24], + } + ), + npartitions=1, + sort=False, + ) + config = { + "translation_rules": { + "DEMO": { + "object_type_translation": { + "conditions": [ + {"expr": "10 <= z_flag < 20", "value": "qso"} + ], + "default": None, + } + } + } + } + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].tolist() == [ + "qso", + pd.NA, + pd.NA, + ] + + +def test_object_type_rejects_values_outside_domain(): + valid = dd.from_pandas( + pd.DataFrame({"object_type_homogenized": ["STAR", "agn", "GALACTIC"]}), + npartitions=1, + sort=False, + ) + result, *_ = _homogenize(valid, {}, "demo", LOGGER, type_cast_ok=False) + assert result.compute()["object_type_homogenized"].tolist() == [ + "star", + "agn", + "galactic", + ] + + invalid = dd.from_pandas( + pd.DataFrame({"object_type_homogenized": ["unknown"]}), + npartitions=1, + sort=False, + ) + with pytest.raises(ValueError, match="Invalid values"): + _homogenize(invalid, {}, "demo", LOGGER, type_cast_ok=False) + + +def test_catalog_object_type_rules_use_renamed_flags_and_conservative_defaults(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": [ + "PRIMUS", + "VVDS", + "VUDS", + "VANDELS", + "VIPERS_PDR2", + "OZDES", + "OZDES", + ], + "object_type": [pd.NA] * 7, + "z_flag": [4, 14.5, 23, 214, 213.5, 4, 6], + "CLASS": ["AGN", pd.NA, pd.NA, pd.NA, pd.NA, pd.NA, pd.NA], + "classFlag": [pd.NA, pd.NA, pd.NA, pd.NA, 1, pd.NA, pd.NA], + "Object_types": [ + pd.NA, + pd.NA, + pd.NA, + pd.NA, + pd.NA, + "Photo-z,LRG", + "QSO_faint", + ], + } + ), + npartitions=2, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "agn", + "agn", + "missing", + "agn", + "agn", + "missing", + "star", + ] + + +def test_3dhst_and_mosdef_object_type_rules_keep_unknowns_null(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["3D-HST", "3D-HST", "MOSDEF", "MOSDEF", "GAMA_DR4"], + "object_type": [pd.NA] * 5, + "z_best_s": [0.0, 1.0, float("nan"), float("nan"), float("nan")], + "TARGET": [float("nan"), float("nan"), 1.0, 0.0, float("nan")], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "star", + "missing", + "missing", + "missing", + "missing", + ] + + +def test_swire_j2_qso_range_stops_at_15(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["SWIRE-REVISED"] * 7, + "object_type": [pd.NA] * 7, + "mst": [0, 0, -1, 1, -5, 5, -5], + "J1": [1, 1, 1, 1, 1, 1, 13], + "J2": [15, 16, 16, 16, 16, 16, 16], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "qso", + "missing", + "star", + "galaxy", + "missing", + "missing", + "qso", + ] + + +def test_2df_6df_and_2mrs_use_only_documented_object_classes(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": [ + "2DFLENS", + "2DFLENS", + "6DFGS", + "2MRS", + "2MRS", + "2MRS", + ], + "object_type": [pd.NA] * 6, + "z_flag": [6, 4, 6, 4, 4, 4], + "TYPE": [pd.NA, pd.NA, pd.NA, "-5A", "-9", "98"], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "galactic", + "missing", + "galactic", + "galaxy", + "agn", + "missing", + ] + + +def test_stellar_classification_does_not_overwrite_independent_quality(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = ["z_flag_homogenized"] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": [ + "2DFLENS", + "6DFGS", + "VIPERS_PDR2", + "ELAISS1OID", + "ELAISS1OID", + "ELAISFBMC", + "ELAISFBMC", + "SWIRE-REVISED", + "SWIRE-REVISED", + "SWIRE-REVISED", + "OZDES", + ], + "z_flag": [6, 6, 4.2, 6, 4, 6, 4, 6, 4, 4, 6], + "classFlag": [0, 0, -1, 0, 0, 0, 0, 0, 0, 0, 0], + "Class": [0, 0, 0, 2, 4, 0, 0, 0, 0, 0, 0], + "tSp": [0, 0, 0, 0, 0, 1, 7, 0, 0, 0, 0], + "mst": [0, 0, 0, 0, 0, 0, 0, 1, -1, -1, 0], + "J1": [1, 1, 1, 1, 1, 1, 1, 1, 1, 13, 1], + "J2": [1] * 11, + } + ), + npartitions=2, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["z_flag_homogenized"].fillna(-1).tolist() == [ + -1, + -1, + 0, + -1, + 4, + -1, + 4, + -1, + 4, + 4, + -1, + ] + + +def test_vimos_uses_only_unambiguous_comm_classifications(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["VIMOS"] * 5, + "object_type": [pd.NA] * 5, + "COMM": [ + "star", + "Star", + "Star?", + "CIV_[CIII]_(BLAGN)", + "Lya(em)_CIV_(BLAGN?)", + ], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "star", + "star", + "missing", + "agn", + "missing", + ] + + +def test_deimos_remarks_use_only_unambiguous_classifications(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["DEIMOS_10K"] * 10, + "object_type": [pd.NA] * 10, + "z_flag": [4] * 8 + [14, 14], + "Remarks": [ + "star", + "M star", + "Star?", + "in halo of bright star", + "NaI,TiO,Ha,star", + "MgII,QSO?", + "CIII],NeIV?(br),QSO", + "MgII(br),MgII(abs),[NeV]br,[OII]br(QSO)", + "-", + "CIV,QSO", + ], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "star", + "star", + "missing", + "missing", + "star", + "missing", + "qso", + "qso", + "agn", + "qso", + ] + + +def test_generic_agn_labels_are_not_promoted_to_qso(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": [ + "HELP-DMU23", + "OZDES", + "OZDES", + "ELAISS1OID", + "ELAISS1OID", + "ELAISFBMC", + "ELAISFBMC", + ], + "object_type": [pd.NA] * 7, + "agn_flag": [1.0] + [float("nan")] * 6, + "Object_types": [ + pd.NA, + "AGN_reverberation", + "AGN_reverberation,XXL_QSO", + pd.NA, + pd.NA, + pd.NA, + pd.NA, + ], + "z_flag": [4.0] * 7, + "Class": [float("nan")] * 3 + [1.0, 5.0] + [float("nan")] * 2, + "tSp": [float("nan")] * 5 + [4.0, 5.0], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].tolist() == [ + "agn", + pd.NA, + pd.NA, + "agn", + "agn", + "agn", + "agn", + ] + + +def test_vipers_keeps_photometric_star_like_null_but_retains_spectroscopic_blagn(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["VIPERS_PDR2"] * 3, + "object_type": [pd.NA] * 3, + "z_flag": [4.2, 13.2, 213.2], + "classFlag": [-1, -1, -1], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "missing", + "agn", + "agn", + ] + + +def test_ozdes_targeting_labels_are_not_treated_as_object_classifications(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame = dd.from_pandas( + pd.DataFrame( + { + "survey": ["OZDES"] * 3, + "object_type": [pd.NA] * 3, + "Object_types": ["RNDstars", "BrightStar", "WhiteDwarf"], + "z_flag": [4] * 3, + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) + + assert result.compute()["object_type_homogenized"].isna().all() + + +def test_euclid_optional_source_is_safe_before_column_arrives(): + config_path = Path(__file__).resolve().parents[1] / "flags_translation.yaml" + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + config["tiebreaking_priority"] = [] + frame_without_source = dd.from_pandas( + pd.DataFrame( + { + "survey": ["EUCLID_Q1"], + "object_type": [pd.NA], + } + ), + npartitions=1, + sort=False, + ) + missing_result, *_ = _homogenize( + frame_without_source, config, "demo", LOGGER, type_cast_ok=False + ) + assert missing_result.compute()["object_type_homogenized"].isna().all() + + frame_with_source = dd.from_pandas( + pd.DataFrame( + { + "survey": ["EUCLID_Q1"] * 5, + "object_type": [pd.NA] * 5, + "spe_class": [pd.NA, "STAR", "GALAXY", "QSO", "UNDEF"], + } + ), + npartitions=1, + sort=False, + ) + + result, *_ = _homogenize( + frame_with_source, config, "demo", LOGGER, type_cast_ok=False + ) + + assert result.compute()["object_type_homogenized"].fillna("missing").tolist() == [ + "missing", + "star", + "galaxy", + "qso", + "missing", + ] + + def test_homogenization_requires_translation_for_every_survey(): frame = dd.from_pandas( pd.DataFrame({"survey": ["known", "missing"], "z_flag": [1, 1]}), @@ -103,7 +862,7 @@ def test_user_homogenized_flag_rejects_values_outside_domain(): _homogenize(frame, config, "demo", LOGGER, type_cast_ok=False) -def test_semantic_star_flag_is_preserved_without_being_a_priority(): +def test_quality_flag_rejects_legacy_six_and_accepts_all_null(): frame = dd.from_pandas( pd.DataFrame( { @@ -116,18 +875,29 @@ def test_semantic_star_flag_is_preserved_without_being_a_priority(): ) config = {"tiebreaking_priority": ["custom_score"]} + with pytest.raises(ValueError, match="Invalid values"): + _homogenize( + frame, + config, + "demo", + LOGGER, + type_cast_ok=False, + require_z_flag_homogenized=True, + ) + + all_null = dd.from_pandas( + pd.DataFrame({"z_flag_homogenized": [pd.NA, pd.NA]}), + npartitions=1, + sort=False, + ) result, *_ = _homogenize( - frame, - config, + all_null, + {"tiebreaking_priority": ["z_flag_homogenized"]}, "demo", LOGGER, type_cast_ok=False, - require_z_flag_homogenized=True, ) - - computed = result.compute() - assert computed["z_flag_homogenized"].tolist() == [4.0, 6.0] - assert "instrument_type_homogenized" not in computed.columns + assert result.compute()["z_flag_homogenized"].isna().all() @pytest.mark.parametrize( @@ -136,7 +906,7 @@ def test_semantic_star_flag_is_preserved_without_being_a_priority(): ("concatenate", 0, False), ("concatenate", None, False), ("concatenate", 3, True), - ("concatenate", "6", True), + ("concatenate", "6", False), ("concatenate", 1.5, False), ("concatenate", 7, False), ("concatenate_and_mark_duplicates", 0, True), @@ -199,7 +969,7 @@ def test_combine_configuration_rejects_invalid_mode_and_empty_dedup_priorities() assert priorities == [] -def test_cut_six_warns_early_for_remove_duplicates(): +def test_cut_six_has_no_legacy_star_semantics(): logger = Mock() validate_combine_configuration( @@ -209,5 +979,4 @@ def test_cut_six_warns_early_for_remove_duplicates(): logger, ) - logger.warning.assert_called_once() - assert "final catalog will normally be empty" in logger.warning.call_args.args[0] + logger.warning.assert_not_called() diff --git a/tests/test_spatial_config.py b/tests/test_spatial_config.py index 71feda6..2b70d83 100644 --- a/tests/test_spatial_config.py +++ b/tests/test_spatial_config.py @@ -9,7 +9,13 @@ def test_spatial_safety_accepts_radius_smaller_than_margin(): - validate_spatial_safety(0.5, 5.0, 0.8) + validate_spatial_safety(0.5, 5.0, 0.8, 1.0) + + +@pytest.mark.parametrize("max_radius", [0.4, 5.0, 6.0]) +def test_spatial_safety_rejects_unsafe_representative_radius(max_radius): + with pytest.raises(ValueError, match="max_representative_radius_arcsec"): + validate_spatial_safety(0.5, 5.0, 0.8, max_radius) @pytest.mark.parametrize( diff --git a/tests/test_tie_treatment.py b/tests/test_tie_treatment.py index 67b1149..f7c2ead 100644 --- a/tests/test_tie_treatment.py +++ b/tests/test_tie_treatment.py @@ -16,7 +16,7 @@ def _results(): return pd.DataFrame( { - "CRD_ID": ["winner", "loser", "tie-a", "tie-b", "star"], + "CRD_ID": ["winner", "loser", "tie-a", "tie-b", "excluded"], "group_id": [1, 1, 2, 2, 3], "tie_result": [1, 0, 2, 2, 3], } @@ -35,7 +35,7 @@ def test_final_tie_filtering_policies(option, expected): result, effective, resolved = filter_pandas_by_tie_treatment(_results(), option) assert set(result["CRD_ID"]) == expected - assert "star" not in set(result["CRD_ID"]) + assert "excluded" not in set(result["CRD_ID"]) assert effective == (option if option != "invalid" else "remove_all") assert resolved == 0 @@ -48,7 +48,7 @@ def test_draw_one_keeps_exactly_one_candidate_per_hard_tie(): assert effective == "draw_one" assert resolved == 1 assert set(result["CRD_ID"]) == {"winner", "tie-a"} - assert "star" not in set(result["CRD_ID"]) + assert "excluded" not in set(result["CRD_ID"]) def test_draw_one_without_group_id_falls_back_to_remove_all():