diff --git a/astra.yaml b/astra.yaml index 2de4a4835..70d7150dd 100644 --- a/astra.yaml +++ b/astra.yaml @@ -323,7 +323,8 @@ analyses: image, and ShapePipe's tile SExtractor run requests no IMAFLAGS_ISO (detection.detection_source_mode). Neighbour pixels are masked separately, from the -1e30 markers SExtractor writes into the tile - VIGNET (shape_measurement.blend_handling). Sky-fixed masks never touch + VIGNET and the coadd segmentation map + (shape_measurement.blend_handling). Sky-fixed masks never touch pixels: a star halo changes no stamp. Values: config_exp_psfex.ini#SEXTRACTOR_RUNNER.FLAG_IMAGE = True; @@ -1567,11 +1568,15 @@ analyses: symmetrized mask as in DES, since symmetrizing quadruples m and leaves an additive c1 (symmetrized_4fold_noise); the hole this leaves near the galaxy is bounded by central_defect_veto. In the - tile VIGNET, SExtractor writes -1e30 over other detections and - beyond the tile edge. Rows and columns that are entirely -1e30 and - start at the stamp border are off-tile, hence defects (the epoch - has the object's light there); every other -1e30 pixel is a - neighbour, handled by blend_handling. + tile VIGNET, SExtractor writes -1e30 beyond the tile edge and over + every above-threshold pixel outside the object's pre-CLEAN + footprint. Rows and columns that are entirely -1e30 and start at + the stamp border are off-tile, hence defects (the epoch has the + object's light there); the other -1e30 pixels on another object's + segmentation footprint are neighbours, handled by blend_handling. + The rest (sub-MINAREA noise peaks, the object's own CLEAN-merged + fragments) are not defects either: the epoch stamps hold real + light there, so they keep their weight and image. Values: VIGNETMAKER_RUNNER_RUN_1.MASKING = False; VIGNETMAKER_RUNNER_RUN_2.MASKING = False; @@ -1638,16 +1643,12 @@ analyses: Covers only pixels shared with a neighbour; defect_fill does not depend on it: defects are filled the same way under every BLEND_HANDLING, and the epoch cuts never count neighbour pixels. - noisefill (the default; the committed config sets no key) gives - weight 0 to the pixels marked -1e30 in the tile VIGNET on other - detections' footprints (about 93% of neighbour-footprint pixels on - a sims tile), excluding the off-tile - rows and columns, which are defects, and replaces them with noise; - unmarked - neighbour pixels keep their weight and light. uberseg ignores the - markers and zeroes the weight of pixels nearer a neighbour's coadd - segmentation footprint than the target's (DILATE_NEIGHBOUR, default 1) - and leaves the image untouched, as official uberseg does: + The committed choice is uberseg (the workflow's default + blend_handling), which ignores the -1e30 neighbour markers in the + tile VIGNET and zeroes the weight of pixels nearer a neighbour's + coadd segmentation footprint than the target's (DILATE_NEIGHBOUR, + default 1), leaving the neighbour side's image untouched, as + official uberseg does: esheldon/meds get_uberseg returns a weight map (a nearest-segment-pixel Voronoi split). DES Y1's fiducial metacal and the last Y3 config ran on uberseg-weighted stamps with raw neighbour @@ -1664,9 +1665,23 @@ analyses: with simulations (mask_des_y1_uberseg_only, mask_blend_bias_detection). Noise-filling the neighbour side would instead cut the target's own light along an unsheared boundary, a - sharp edge that rings in the FFTs. The recommended comparison arm is - uberseg (weight-only); defect_fill is the same in both arms. - default: noisefill + sharp edge that rings in the FFTs. The alternative, noisefill + (ngmix's own default, and the workflow's blend_handling: + noisefill), gives weight 0 to the pixels marked -1e30 in the tile + VIGNET (about 93% of neighbour-footprint pixels on a simulated + tile), excluding the off-tile rows and columns, which are defects, + and replaces them with noise; unmarked neighbour pixels keep their + weight and light. Under both handlings the neighbours are defined + by the coadd segmentation map: a marker counts only on a footprint + labelled neither 0 nor the target. SExtractor's MASK_TYPE CORRECT + also marks sub-MINAREA sky blobs and fragments CLEAN merged into + the target, which the seg labels as sky or as the target; noise- + filling those removed the target's own light (15% of + NEIGHBOUR_FLAG = 0 objects differed between the arms on 25 tiles). + tile_detect therefore writes SEG_VIGNET under both handlings. + Uberseg reads no marker, so its output is unchanged. It is + the comparison arm, with defect_fill held equal across arms. + default: uberseg options: noisefill: label: Noise-fill the marked neighbour pixels (BLEND_HANDLING = noisefill) @@ -1681,8 +1696,8 @@ analyses: uberseg: label: UberSeg, weight-only description: >- - As in DES Y1/Y3. Needs the coadd segmentation stamp - (SEG_VIGNET_PATH). DILATE_NEIGHBOUR absorbs the coadd-vs-epoch + As in DES Y1/Y3. Needs the coadd segmentation stamp (the tile + catalogue's SEG_VIGNET). DILATE_NEIGHBOUR absorbs the coadd-vs-epoch overlay offset: ShapePipe reuses one coadd seg stamp for every epoch where MEDS reprojects it. insights: [mask_uberseg_weight_only, mask_uberseg_neighbour_bias, mask_des_y1_uberseg_only, mask_blend_bias_detection] diff --git a/src/shapepipe/modules/make_cat_package/make_cat.py b/src/shapepipe/modules/make_cat_package/make_cat.py index c289b4d65..7be609147 100644 --- a/src/shapepipe/modules/make_cat_package/make_cat.py +++ b/src/shapepipe/modules/make_cat_package/make_cat.py @@ -92,7 +92,8 @@ def read_sextractor_data(sexcat_path, remove_vignet=True): sexcat_path : str Path to SExtractor catalogue remove_vignet : bool - If ``True`` will not keep the ``VIGNET`` field + If ``True`` will not keep the ``VIGNET`` and ``SEG_VIGNET`` stamp + fields Returns ------- @@ -109,7 +110,7 @@ def read_sextractor_data(sexcat_path, remove_vignet=True): columns = { name: data[name] for name in data.dtype.names - if not (remove_vignet and name == "VIGNET") + if not (remove_vignet and name in ("VIGNET", "SEG_VIGNET")) } tile_name = os.path.basename(sexcat_path) diff --git a/src/shapepipe/modules/ngmix_package/ngmix.py b/src/shapepipe/modules/ngmix_package/ngmix.py index bf1cc86fb..43995f926 100644 --- a/src/shapepipe/modules/ngmix_package/ngmix.py +++ b/src/shapepipe/modules/ngmix_package/ngmix.py @@ -589,13 +589,10 @@ class Tile_cat(): Parameters ---------- cat_path : str - Path to the tile SExtractor catalogue. - seg_cat_path : str, optional - Path to the coadd-frame segmentation VIGNET catalogue (a CLASSIC-mode - vignetmaker output cut from the tile ``SEGMENTATION`` check image), - row-aligned to ``cat_path``. When given, ``self.seg`` holds one integer - seg stamp per object for the ``"uberseg"`` blend handling; ``None`` - leaves ``self.seg`` unset, which only ``"uberseg"`` needs. + Path to the tile SExtractor catalogue. Its optional ``SEG_VIGNET`` + column, one integer coadd segmentation stamp per object on the grid + of its ``VIGNET``, becomes ``self.seg``, which defines the neighbours + under both blend handlings; without it ``self.seg`` is ``None``. row_min, row_max : int, optional First and last catalogue row of the chunk (1-based, inclusive; see :func:`chunk_rows`). The default, ``-1``, is unbounded, so the whole @@ -605,12 +602,10 @@ class Tile_cat(): def __init__( self, cat_path, - seg_cat_path=None, row_min=-1, row_max=-1, ): self.cat_path = cat_path - self.seg_cat_path = seg_cat_path self.row_min = row_min self.row_max = row_max if cat_path: @@ -639,41 +634,17 @@ def get_data(self, cat_path): else None ) - tile_cat.close() + # Coadd-frame segmentation stamp (integer labels, the catalogue's + # NUMBER), one per object on the grid of its VIGNET, overlaid + # unchanged on every epoch; it defines the neighbours (shapepipe#776, + # split_tile_markers, uberseg_mask). + self.seg = ( + ChunkStamps(data['SEG_VIGNET'], self.rows) + if 'SEG_VIGNET' in cols + else None + ) - # Coadd-frame SExtractor segmentation stamp (integer labels), one per - # object and row-aligned to the tile catalogue, overlaid unchanged on - # every epoch for uberseg neighbour masking (shapepipe#776). None -> - # uberseg unavailable; ``"noisefill"`` does not read it. - self.seg = None - if self.seg_cat_path: - seg_cat = file_io.FITSCatalogue( - self.seg_cat_path, - SEx_catalogue=True, - memmap=True, - ) - seg_cat.open() - seg_data = seg_cat.get_data() - # The seg VIGNETs are indexed by tile-catalogue row (self.seg[i]), - # so the two catalogues MUST be row-aligned. Fail loud at load if - # they are not — a silent length/order mismatch would hand every - # object the wrong footprint and quietly corrupt every mask. - if len(seg_data) != len(self.obj_id): - raise ValueError( - f"SEG_VIGNET_PATH '{self.seg_cat_path}' has" - + f" {len(seg_data)} rows but the tile catalogue has" - + f" {len(self.obj_id)}; the segmentation vignets must be" - + " row-aligned to the tile catalogue." - ) - if 'NUMBER' in seg_data.dtype.names: - if not np.array_equal(seg_data['NUMBER'], self.obj_id): - raise ValueError( - f"SEG_VIGNET_PATH '{self.seg_cat_path}' NUMBER column" - + " does not match the tile catalogue NUMBER; the" - + " segmentation vignets are misaligned or reordered." - ) - self.seg = ChunkStamps(seg_data['VIGNET'], self.rows) - seg_cat.close() + tile_cat.close() class Postage_stamp(): """Galaxy Postage Stamp. @@ -871,13 +842,11 @@ class Ngmix(object): blend_handling : {"noisefill", "uberseg"}, optional Neighbour treatment. ``"noisefill"`` (default) zero-weights and noise-fills the pixels marked -1e30 in the tile VIGNET on other - detections' footprints; ``"uberseg"`` ignores those markers, + detections' footprints; ``"uberseg"`` ignores those markers and zeroes the weight of neighbour-side pixels from the coadd - segmentation map and requires ``seg_cat_path``. Defect pixels are - filled under both (see :func:`prepare_ngmix_weights`). - seg_cat_path : str, optional - Path to the coadd-frame segmentation VIGNET catalogue (see - :class:`Tile_cat`). Required when ``blend_handling="uberseg"``. + segmentation stamps, the tile catalogue's ``SEG_VIGNET`` column (see + :class:`Tile_cat`), which it requires. Defect pixels are filled + under both (see :func:`prepare_ngmix_weights`). dilate_neighbour : int, optional Neighbour-mask dilation iterations for ``"uberseg"`` (see :func:`uberseg_mask`); the default is ``1``. @@ -902,7 +871,7 @@ class Ngmix(object): If the length of the input file list is incorrect ValueError If ``blend_handling``, ``defect_weighting`` or ``defect_fill`` is - unknown, or ``"uberseg"`` is selected without ``seg_cat_path``. + unknown. """ @@ -920,7 +889,6 @@ def __init__( bkg_sub=True, centroid_source="wcs", blend_handling="noisefill", - seg_cat_path=None, dilate_neighbour=1, metacal_psf="fitgauss", defect_weighting=DEFECT_WEIGHTING, @@ -956,14 +924,6 @@ def __init__( + f" {DEFECT_FILLS}" ) - # Fail fast at construction (not deep in the per-epoch loop) when - # uberseg is requested without its segmentation input (shapepipe#776). - if blend_handling == "uberseg" and seg_cat_path is None: - raise ValueError( - "blend_handling='uberseg' requires SEG_VIGNET_PATH (the coadd" - + " SExtractor segmentation vignets); none configured." - ) - self._tile_cat_path = input_file_list[0] if bkg_sub: bkg_path, psf_path, weight_path, flag_path = ( @@ -1003,7 +963,6 @@ def __init__( self._bkg_sub = bkg_sub self._centroid_source = centroid_source self._blend_handling = blend_handling - self._seg_cat_path = seg_cat_path self._dilate_neighbour = dilate_neighbour self._metacal_psf = metacal_psf self._defect_weighting = defect_weighting @@ -1400,12 +1359,27 @@ def process(self): """ tile_cat = Tile_cat( self._tile_cat_path, - self._seg_cat_path, self._id_obj_min, self._id_obj_max, ) vignet_cat = self._vignet_cat + # Fail before the per-object loop, whose try/except would otherwise + # drop every object one by one (shapepipe#776). + if self._blend_handling == "uberseg" and tile_cat.seg is None: + raise ValueError( + "BLEND_HANDLING = uberseg needs the tile catalogue's" + + f" SEG_VIGNET column, which {self._tile_cat_path} lacks;" + + " write it at tile detection (SEG_VIGNET = True)." + ) + if tile_cat.seg is None and getattr(tile_cat, "vign", None) is not None: + self._w_log.warning( + f"{self._tile_cat_path} has no SEG_VIGNET column: every" + + " interior -1e30 marker is taken as a neighbour," + + " including the object's own CLEAN-merged fragments" + + " (neighbours-are-other-footprints)." + ) + check_wcs_centroid_offset( self._centroid_source, tile_cat, vignet_cat.gal_vign_cat ) @@ -1529,7 +1503,7 @@ def process(self): res['obj_id'] = obj_id # Neighbour flag: does the coadd seg stamp hold any non-central, # non-zero label? Systematics-test hook (shapepipe#776); computed - # from the raw coadd seg, 0 when uberseg is off / seg absent. + # from the raw coadd seg, 0 when the seg is absent. res['neighbour_flag'] = int( seg_has_neighbour(tile_cat.seg[i_tile], obj_id) if getattr(tile_cat, "seg", None) is not None @@ -1630,8 +1604,8 @@ def prepare_postage_stamps( that ``DEFECT_WEIGHTING`` zero-weights or interpolates are not counted. @sc [decision:shape_measurement.central_defect_veto,decision:shape_measurement.epoch_masked_fraction_cut,decision:shape_measurement.blend_handling] neighbour-markers-are-not-defects - The tile VIGNET's -1e30 markers on other detections' footprints - (:func:`split_tile_markers`) are the epoch's neighbour mask + The tile VIGNET's -1e30 markers on other objects' segmentation + footprints (:func:`split_tile_markers`) are the epoch's neighbour mask (``stamp.neighbours``), not defects: every epoch shares the tile VIGNET, so counting them would drop every epoch of a blended object. @@ -1738,7 +1712,9 @@ def prepare_postage_stamps( flag_vign = flag_obj[expccd_name]['VIGNET'] # Off-tile pixels are defects (off-tile-pixels-are-defects); the # other -1e30 markers are neighbours (neighbour-markers-are-not-defects). - neighbour, off_tile = split_tile_markers(tile_vign, np.shape(gal_vign)) + neighbour, off_tile = split_tile_markers( + tile_vign, np.shape(gal_vign), tile_seg, obj_id + ) weight_vign = np.where(off_tile, 0, weight_obj[expccd_name]['VIGNET']) bkg_rms_vign = ( bkg_rms_obj[expccd_name]['VIGNET'] @@ -1812,18 +1788,28 @@ def prepare_postage_stamps( return stamp -def split_tile_markers(tile_vign, shape): +def split_tile_markers(tile_vign, shape, seg=None, object_number=None): """Split the tile VIGNET's -1e30 markers into neighbour and off-tile. @sc [decision:shape_measurement.blend_handling,decision:shape_measurement.defect_fill] off-tile-is-marked-border-rows-and-columns - The tile VIGNET holds -1e30 on the footprints of other detections and on - stamp pixels beyond the tile's edge (SExtractor writes it). A stamp - clipped by the tile's rectangle loses whole rows and whole columns from - its border, so the off-tile pixels are the union of the runs of - entirely -1e30 rows and columns that start at a stamp border. The remaining markers are - neighbour pixels: a footprint touching the stamp border, and a footprint - that completes an interior row or column beside an off-tile band, stay - neighbours. + SExtractor writes -1e30 into the tile VIGNET on stamp pixels beyond the + tile's edge and, under MASK_TYPE CORRECT, on every above-threshold pixel + that is not the object's own pre-CLEAN footprint. A stamp clipped by the + tile's rectangle loses whole rows and whole columns from its border, so + the off-tile pixels are the union of the runs of entirely -1e30 rows and + columns that start at a stamp border. + + @sc [decision:shape_measurement.blend_handling] neighbours-are-other-footprints + The other markers also fall on pixels no other object owns: detections + smaller than DETECT_MINAREA (noise peaks, catalogued nowhere) and + fragments CLEAN merged into the object itself, which the segmentation + map labels as the object. With ``seg``, the neighbour pixels are the + markers on another label's footprint (any label but 0 and + ``object_number``; the catalogue join's -1 marks a detection that left + the catalogue). The markers left over are nothing: -1e30 is a flag in the + tile VIGNET only, and the epoch stamps hold real light there, the + object's own or sky. Without ``seg``, every interior marker is a + neighbour. Parameters ---------- @@ -1831,6 +1817,10 @@ def split_tile_markers(tile_vign, shape): Tile VIGNET stamp, oriented like the epoch; ``None`` marks nothing. shape : tuple of int Stamp shape, used when ``tile_vign`` is ``None``. + seg : numpy.ndarray, optional + The object's SEG_VIGNET stamp, oriented like ``tile_vign``. + object_number : int, optional + The object's label in ``seg``; required with ``seg``. Returns ------- @@ -1853,7 +1843,10 @@ def border_runs(full): border_runs(marker.all(axis=1))[:, None] | border_runs(marker.all(axis=0))[None, :] ) - return marker & ~off_tile, off_tile + neighbour = marker & ~off_tile + if seg is not None: + neighbour &= (seg != 0) & (seg != object_number) + return neighbour, off_tile def background_subtract(gal,bkg): @@ -2480,7 +2473,7 @@ def prepare_ngmix_weights( raise ValueError( "blend_handling='uberseg' requires a segmentation map and the" + " central object_number; none reached prepare_ngmix_weights." - + " Set SEG_VIGNET_PATH on the ngmix run (see" + + " The tile catalogue's SEG_VIGNET column carries the map (see" + " CosmoStat/shapepipe#776)." ) diff --git a/src/shapepipe/modules/ngmix_runner.py b/src/shapepipe/modules/ngmix_runner.py index 44e7add37..d55f19c06 100644 --- a/src/shapepipe/modules/ngmix_runner.py +++ b/src/shapepipe/modules/ngmix_runner.py @@ -91,24 +91,6 @@ def ngmix_runner( else: input_file_list = input_file_list[:wcs_idx] - # SEG_VIGNET_PATH (optional): coadd-frame SExtractor segmentation vignets - # (a CLASSIC-mode vignetmaker output), row-aligned to the tile catalogue. - # Required for BLEND_HANDLING = uberseg; when set, the file must exist for - # every tile (missing file -> error). Read on Tile_cat, not via Vignet, so - # it is threaded to Ngmix as its own argument rather than into - # input_file_list. - if config.has_option(module_config_sec, "SEG_VIGNET_PATH"): - seg_vignet_path = config.getexpanded( - module_config_sec, - "SEG_VIGNET_PATH", - ).format(file_number_string=file_number_string) - if not os.path.exists(seg_vignet_path): - raise FileNotFoundError( - f"Segmentation vignet file not found: {seg_vignet_path}" - ) - else: - seg_vignet_path = None - # Batch save option if config.has_option(module_config_sec, "SAVE_BATCH"): save_batch = config.getint( @@ -136,15 +118,16 @@ def ngmix_runner( centroid_source = "wcs" # Neighbour treatment: "noisefill" (default) zero-weights and noise-fills - # the pixels marked -1e30 in the tile VIGNET (other detections' - # footprints); "uberseg" ignores those markers, zeroes the weight of every + # the pixels marked -1e30 in the tile VIGNET on other objects' + # segmentation footprints (split_tile_markers); "uberseg" ignores those + # markers, zeroes the weight of every # pixel closer to a neighbour than to the central object, from the # segmentation map, and leaves its image raw. Defect pixels (flagged, # zero-weight, invalid-RMS or off-tile) are zero-weighted and filled the # same way under both, and the epoch cuts are fixed; see # prepare_postage_stamps and prepare_ngmix_weights. if config.has_option(module_config_sec, "BLEND_HANDLING"): - blend_handling = config.get(module_config_sec, "BLEND_HANDLING") + blend_handling = config.getexpanded(module_config_sec, "BLEND_HANDLING") else: blend_handling = "noisefill" @@ -238,7 +221,6 @@ def ngmix_runner( bkg_sub=bkg_sub, centroid_source=centroid_source, blend_handling=blend_handling, - seg_cat_path=seg_vignet_path, dilate_neighbour=dilate_neighbour, metacal_psf=metacal_psf, defect_weighting=defect_weighting, diff --git a/src/shapepipe/modules/sextractor_package/match_catalogue.py b/src/shapepipe/modules/sextractor_package/match_catalogue.py index 68bca78f7..6c59d31e7 100644 --- a/src/shapepipe/modules/sextractor_package/match_catalogue.py +++ b/src/shapepipe/modules/sextractor_package/match_catalogue.py @@ -23,9 +23,13 @@ # The SEG_VIGNET label of a footprint whose SExtractor row has no partner in # the external catalogue, and so leaves the catalogue. Negative, so it never -# collides with a NUMBER; UberSeg only asks "self or not self". +# collides with a NUMBER (match_catalogue requires positive ones); UberSeg only +# asks "self or not self". UNMATCHED_LABEL = -1 +# The largest NUMBER the int32 NUMBER and SEG_VIGNET columns hold. +MAX_NUMBER = np.iinfo(np.int32).max + def mutual_nearest(x_a, y_a, x_b, y_b, radius): """One-to-one pairs of mutual nearest neighbours closer than ``radius``. @@ -100,9 +104,12 @@ def match_catalogue(cat_path, ext_cat_path, radius=1.0, min_fraction=0.98, external catalogue (``X_IMAGE``, ``Y_IMAGE``, same image grid) within ``radius`` pixels. Paired rows take the external ``NUMBER``; unpaired rows leave the catalogue. A ``SEG_VIGNET`` column, when present, is - relabelled to the new numbering, with the footprints of rows that left - marked ``UNMATCHED_LABEL``. The catalogue is rewritten in place; every - other HDU and column is kept. + relabelled through the whole map from old to new numbers, with the + footprints of rows that left marked ``UNMATCHED_LABEL``: as the external + numbers are unique integers in ``[1, MAX_NUMBER]``, each row's own + footprint carries its new ``NUMBER`` and no other footprint can, + whatever the two numberings share. + The catalogue is rewritten in place; every other HDU and column is kept. Both catalogues come from the same pixels, so pairs agree to ~1e-4 pixel, and on eight DR6 tiles at least 99.2% of each side pairs; the @@ -144,13 +151,25 @@ def match_catalogue(cat_path, ext_cat_path, radius=1.0, min_fraction=0.98, Raises ------ ValueError - If more than ``tolerated_unpaired`` rows of either side, and more - than ``1 - min_fraction`` of it, have no partner + If the external ``NUMBER`` repeats or is not an integer in + ``[1, MAX_NUMBER]``, or if more + than ``tolerated_unpaired`` rows of either side, and more than + ``1 - min_fraction`` of it, have no partner @sc [decision:detection.tile_detection] """ ext = asc.read(ext_cat_path, format="sextractor", include_names=["NUMBER", "X_IMAGE", "Y_IMAGE"]) + ext_number = np.asarray(ext["NUMBER"]) + if (not np.issubdtype(ext_number.dtype, np.integer) + or (ext_number <= 0).any() or (ext_number > MAX_NUMBER).any() + or len(np.unique(ext_number)) < len(ext)): + raise ValueError( + f"{ext_cat_path} has a NUMBER that is not an integer in" + + f" [1, {MAX_NUMBER}] or that repeats; the join needs unique" + + " numbers that fit the int32 NUMBER and SEG_VIGNET columns, so" + + " that no relabelled footprint takes another object's number." + ) with fits.open(cat_path) as hdul: hdus = [hdu.copy() for hdu in hdul] objects = next(h for h in hdus if h.name == "LDAC_OBJECTS") @@ -194,7 +213,7 @@ def too_many_unpaired(n_total): old_number = np.asarray(data["NUMBER"]) new_number = np.full(len(data), UNMATCHED_LABEL, np.int64) - new_number[i_sex] = np.asarray(ext["NUMBER"])[i_ext] + new_number[i_sex] = ext_number[i_ext] columns = [] for col in objects.columns: diff --git a/src/shapepipe/modules/sextractor_package/sextractor_script.py b/src/shapepipe/modules/sextractor_package/sextractor_script.py index 002bf14f7..e41ddad57 100644 --- a/src/shapepipe/modules/sextractor_package/sextractor_script.py +++ b/src/shapepipe/modules/sextractor_package/sextractor_script.py @@ -16,6 +16,190 @@ from shapepipe.pipeline.sqlite_store import read_sqlitedict +def cut_stamps(array, col, row, stamp_size, fill): + """Cut one square stamp per object from a 2-D array. + + Each stamp is ``stamp_size`` pixels on a side with the 0-based pixel + (``row``, ``col``) at index ``stamp_size // 2`` on both axes; pixels off + the array take ``fill``. This is the window SExtractor cuts ``VIGNET`` + with, so stamps cut at the same centres from arrays on one pixel grid are + registered pixel for pixel. + + Parameters + ---------- + array : numpy.ndarray + 2-D array, shape ``(ny, nx)`` + col, row : array_like of int + 0-based column and row of each stamp's centre pixel + stamp_size : int + Side length of the stamps + fill : scalar + Value of the stamp pixels off the array + + Returns + ------- + numpy.ndarray + Stamps, shape ``(n_obj, stamp_size, stamp_size)``, dtype of ``array`` + + """ + ny, nx = array.shape + half = stamp_size // 2 + stamps = np.full((len(col), stamp_size, stamp_size), fill, array.dtype) + for i, (xi, yi) in enumerate(zip(col, row)): + x0, y0 = xi - half, yi - half + xc0, xc1 = max(0, x0), min(nx, x0 + stamp_size) + yc0, yc1 = max(0, y0), min(ny, y0 + stamp_size) + if xc0 >= xc1 or yc0 >= yc1: + continue + stamps[i, yc0 - y0:yc1 - y0, xc0 - x0:xc1 - x0] = ( + array[yc0:yc1, xc0:xc1] + ) + return stamps + + +def seg_vignet_column(seg_vignets): + """The ``SEG_VIGNET`` LDAC column: one int32 seg stamp per object. + + Parameters + ---------- + seg_vignets : numpy.ndarray + Segmentation stamps, shape ``(n_obj, stamp_size, stamp_size)``, cut + on ``VIGNET``'s grid (:func:`cut_stamps`) + + Returns + ------- + astropy.io.fits.Column + The column, laid out as ``VIGNET`` is + + """ + n_obj, ny, nx = seg_vignets.shape + return fits.Column( + name="SEG_VIGNET", + format=f"{ny * nx}J", + array=seg_vignets.astype(np.int32, copy=False).reshape(n_obj, -1), + dim=f"({nx},{ny})", + ) + + +# Double-precision positions add_seg_vignet centres SEG_VIGNET on. +DOUBLE_POSITIONS = ("X_IMAGE_DBL", "Y_IMAGE_DBL") + + +def vignet_centre(pos): + """The 0-based pixel SExtractor centres VIGNET on, along one axis. + + SExtractor 2.25.0 (``src/analyse.c``, ``ix=(int)(obj->mx+0.49999)``, + unchanged in Debian's 2.25.0+ds-3) truncates the 0-based barycentre plus + 0.49999. That is not round-to-nearest: a fractional part in + [0.5, 0.50001) goes down, where ``np.rint`` would go up (or to even). + + Parameters + ---------- + pos : array_like of float + 1-based double-precision position (``X_IMAGE_DBL`` or + ``Y_IMAGE_DBL``; SExtractor writes ``mx + 1``) + + Returns + ------- + numpy.ndarray + 0-based pixel index, int64 + + """ + mx = np.asarray(pos, np.float64) - 1.0 + return np.trunc(mx + 0.49999).astype(np.int64) + + +def seg_vignet_param_file(dot_param, output_path): + """Write a SExtractor parameter file that also asks for the double + positions. + + SExtractor centres VIGNET from its double-precision barycentre + (:func:`vignet_centre`), which the float32 ``X_IMAGE`` / ``Y_IMAGE`` + cannot resolve near half pixels; ``X_IMAGE_DBL`` / ``Y_IMAGE_DBL`` + carry it. + :func:`add_seg_vignet` reads them and drops them again. + + Parameters + ---------- + dot_param : str + Path to the configured parameter file + output_path : str + Path to write the extended parameter file to + + Returns + ------- + str + ``output_path`` + + """ + with open(dot_param) as f: + text = f.read() + if text and not text.endswith("\n"): + text += "\n" + with open(output_path, "w") as f: + f.write(text + "".join(f"{name}\n" for name in DOUBLE_POSITIONS)) + return output_path + + +def add_seg_vignet(cat_path, seg_path, w_log=None): + """Add the ``SEG_VIGNET`` column to a SExtractor catalogue. + + The SEGMENTATION check image, whose labels are the catalogue's + ``NUMBER``, is cut on the grid of each object's VIGNET, centred where + SExtractor centres VIGNET (:func:`vignet_centre` of ``X_IMAGE_DBL`` / + ``Y_IMAGE_DBL``, :func:`cut_stamps`), and written, int32 and 0 off the image, as + ``SEG_VIGNET`` in ``LDAC_OBJECTS``, which ngmix's UberSeg blend handling + reads. The double positions (:func:`seg_vignet_param_file`) are dropped; + every other HDU and column is kept. + + Parameters + ---------- + cat_path : str + Path to the SExtractor FITS-LDAC catalogue, rewritten in place + seg_path : str + Path to the SEGMENTATION check image + w_log : logging.Logger, optional + Pipeline logger + + Raises + ------ + ValueError + If the catalogue lacks the double positions + + """ + with fits.open(cat_path) as hdul: + hdus = [hdu.copy() for hdu in hdul] + objects = next(h for h in hdus if h.name == "LDAC_OBJECTS") + data = objects.data + missing = [name for name in DOUBLE_POSITIONS if name not in data.names] + if missing: + raise ValueError( + f"{cat_path} lacks {', '.join(missing)}, which SEG_VIGNET is" + + " centred on; run SExtractor with seg_vignet_param_file." + ) + col = vignet_centre(data["X_IMAGE_DBL"]) + row = vignet_centre(data["Y_IMAGE_DBL"]) + size = data["VIGNET"].shape[1] + seg_vignets = cut_stamps(fits.getdata(seg_path), col, row, size, 0) + kept = [c for c in objects.columns.columns + if c.name not in DOUBLE_POSITIONS] + new = fits.BinTableHDU.from_columns( + kept + [seg_vignet_column(seg_vignets)], + header=objects.header, + name="LDAC_OBJECTS", + ) + fits.HDUList( + [new if h is objects else h for h in hdus] + ).writeto(cat_path, overwrite=True) + if w_log: + centre = size // 2 + own = np.mean(seg_vignets[:, centre, centre] == data["NUMBER"]) + w_log.info( + f"SEG_VIGNET cut from {seg_path} for {len(col)} objects;" + + f" centre label = NUMBER for {own:.4f}" + ) + + def get_header_value(image_path, key): """Get Header Value. @@ -154,12 +338,9 @@ def make_post_process(cat_path, f_wcs_path, pos_params, ccd_size, w_log=None): If SQL file not found """ - cat = file_io.FITSCatalogue( - cat_path, - SEx_catalogue=True, - open_mode=file_io.BaseCatalogue.OpenMode.ReadWrite, - ) - cat.open() + with fits.open(cat_path) as hdul: + hdus = [hdu.copy() for hdu in hdul] + objects = next(h for h in hdus if h.name == "LDAC_OBJECTS") # One lock-free read of the whole header log: merge_headers wrote and # closed it in an earlier step, and keyed SqliteDict reads would take an @@ -174,7 +355,7 @@ def make_post_process(cat_path, f_wcs_path, pos_params, ccd_size, w_log=None): n_hdu = len(f_wcs[exp_keys[0]]) history = [] - for idx in cat.get_data(1)[0][0]: + for idx in hdus[1].data[0][0]: if re.split("HISTORY", idx)[0] == "": history.append(idx) @@ -190,11 +371,13 @@ def make_post_process(cat_path, f_wcs_path, pos_params, ccd_size, w_log=None): exp_list = list(dict.fromkeys(exp_list)) - obj_id = np.copy(cat.get_data()["NUMBER"]) + obj_id = np.copy(objects.data["NUMBER"]) - ra = np.copy(cat.get_data()[pos_params[0]]) - dec = np.copy(cat.get_data()[pos_params[1]]) + ra = np.copy(objects.data[pos_params[0]]) + dec = np.copy(objects.data[pos_params[1]]) + column = file_io.FITSCatalogue.fits_column + epochs = [] n_epoch = np.zeros(len(obj_id), dtype="int32") for idx, exp in enumerate(exp_list): if exp not in f_wcs: @@ -234,21 +417,24 @@ def make_post_process(cat_path, f_wcs_path, pos_params, ccd_size, w_log=None): ind[np.where(near)[0][ind_near]] = True pos_tmp[ind] = idx_j n_epoch[ind] += 1 - exp_name = np.array([exp_list[idx] for n in range(len(obj_id))]) - a = np.array( - [(obj_id[ii], exp_name[ii], pos_tmp[ii]) for ii in range(len(exp_name))], - dtype=[ - ("NUMBER", obj_id.dtype), - ("EXP_NAME", exp_name.dtype), - ("CCD_N", pos_tmp.dtype), + epochs.append(fits.BinTableHDU.from_columns( + [ + column("NUMBER", obj_id), + column("EXP_NAME", np.full(len(obj_id), exp)), + column("CCD_N", pos_tmp), ], - ) - cat.save_as_fits(data=a, ext_name=f"EPOCH_{idx}") - cat.open() - - cat.add_col("N_EPOCH", n_epoch) + name=f"EPOCH_{idx}", + )) - cat.close() + # One write of the whole catalogue: LDAC_OBJECTS with N_EPOCH appended + # (laid out as FITSCatalogue.add_col lays it out), then the EPOCH HDUs. + new = fits.BinTableHDU.from_columns( + objects.data.columns + fits.ColDefs([column("N_EPOCH", n_epoch)]), + name="LDAC_OBJECTS", + ) + fits.HDUList( + [new if h is objects else h for h in hdus] + epochs + ).writeto(cat_path, overwrite=True) class SExtractorCaller: @@ -536,6 +722,7 @@ def get_check_image(self, check_image): if (len(check_image) == 1) & (check_image[0] == ""): check_type = ["NONE"] check_name = ["none"] + self.check_paths = {} else: check_type = [] check_name = [] @@ -549,6 +736,7 @@ def get_check_image(self, check_image): + self._num_str + ".fits" ) + self.check_paths = dict(zip(check_type, check_name)) self._cmd_line_extra += ( f' -CHECKIMAGE_TYPE {",".join(check_type)} ' diff --git a/src/shapepipe/modules/sextractor_runner.py b/src/shapepipe/modules/sextractor_runner.py index 65181712f..2a0df4713 100644 --- a/src/shapepipe/modules/sextractor_runner.py +++ b/src/shapepipe/modules/sextractor_runner.py @@ -6,7 +6,11 @@ """ +import contextlib +import os import re +import shutil +import tempfile from shapepipe.modules.module_decorator import module_runner from shapepipe.modules.sextractor_package import match_catalogue as mc @@ -79,70 +83,122 @@ def sextractor_runner( f_wcs_path = input_file_list[-1] input_file_list = list(input_file_list[:-1]) - # Create sextractor caller class instance - ss_inst = ss.SExtractorCaller( - input_file_list, - run_dirs["output"], - file_number_string, - dot_sex, - dot_param, - dot_conv, - weight_file, - flag_file, - psf_file, - detection_image, - detection_weight, - zp_from_header, - bkg_from_header, - zero_point_key=zp_key, - background_key=bkg_key, - check_image=check_image, - output_prefix=prefix, - ) - - # Generate sextractor command line - command_line = ss_inst.make_command_line(exec_path) - w_log.info(f"Calling command: {command_line}") - - # Execute command line - stderr, stdout = execute(command_line) - - # Parse SExtractor errors - stdout, stderr = ss_inst.parse_errors(stderr, stdout) - - # MATCH_CATALOGUE (optional, environment-expanded path; empty for none): - # take membership and NUMBER from that external catalogue of the same - # image, before the post-processing keys the epoch HDUs on NUMBER. - match_path = ( - config.getexpanded(module_config_sec, "MATCH_CATALOGUE") - if config.has_option(module_config_sec, "MATCH_CATALOGUE") - else "" - ) - if match_path: - mc.match_catalogue( - ss_inst.path_output_file, - match_path, - radius=config.getfloat(module_config_sec, "MATCH_RADIUS"), - min_fraction=config.getfloat( - module_config_sec, "MATCH_MIN_FRACTION" - ), - tolerated_unpaired=config.getint( - module_config_sec, "MATCH_TOLERATED_UNPAIRED" - ), - w_log=w_log, + # SEG_VIGNET (optional, environment-expanded boolean): add the + # SEGMENTATION check image's stamps, on each VIGNET's grid, as the + # SEG_VIGNET column ngmix's UberSeg blend handling reads. SExtractor then + # also writes the double-precision positions VIGNET is centred on, which + # add_seg_vignet reads and drops. + seg_vignet = config.has_option( + module_config_sec, "SEG_VIGNET" + ) and config.getexpandedboolean(module_config_sec, "SEG_VIGNET") + if seg_vignet: + if "SEGMENTATION" not in [key.upper() for key in check_image]: + raise ValueError( + "SEG_VIGNET needs the SEGMENTATION check image in CHECKIMAGE." + ) + dot_param = ss.seg_vignet_param_file( + dot_param, + f"{run_dirs['tmp']}/seg_vignet{file_number_string}.param", ) - # Run sextractor post processing - if config.getboolean(module_config_sec, "MAKE_POST_PROCESS"): - pos_params = config.getlist(module_config_sec, "WORLD_POSITION") - ccd_size = config.getlist(module_config_sec, "CCD_SIZE") - ss.make_post_process( - ss_inst.path_output_file, - f_wcs_path, - pos_params, - ccd_size, - w_log=w_log, + # WORK_DIR (optional, environment-expanded): SExtractor writes its + # catalogue and check images in a fresh directory there, the SEG_VIGNET + # cut, the join and the post-processing each rewrite the catalogue there, + # and the files are moved to the run's output directory once complete. + # On node-local disk this keeps the whole-catalogue rewrites off NFS, + # where each costs minutes under load. The directory is removed on the way + # out, also on failure. Without WORK_DIR everything is written in the + # output directory. + if config.has_option(module_config_sec, "WORK_DIR"): + work = tempfile.TemporaryDirectory( + prefix=f"sp-detect{file_number_string}.", + dir=config.getexpanded(module_config_sec, "WORK_DIR"), + ) + else: + work = contextlib.nullcontext(run_dirs["output"]) + + with work as work_dir: + # Create sextractor caller class instance + ss_inst = ss.SExtractorCaller( + input_file_list, + work_dir, + file_number_string, + dot_sex, + dot_param, + dot_conv, + weight_file, + flag_file, + psf_file, + detection_image, + detection_weight, + zp_from_header, + bkg_from_header, + zero_point_key=zp_key, + background_key=bkg_key, + check_image=check_image, + output_prefix=prefix, + ) + + # Generate sextractor command line + command_line = ss_inst.make_command_line(exec_path) + w_log.info(f"Calling command: {command_line}") + + # Execute command line + stderr, stdout = execute(command_line) + + # Parse SExtractor errors + stdout, stderr = ss_inst.parse_errors(stderr, stdout) + + # SEG_VIGNET is cut before the join, which relabels its stamps with the + # rows' new NUMBERs. + if seg_vignet: + ss.add_seg_vignet( + ss_inst.path_output_file, + ss_inst.check_paths["SEGMENTATION"], + w_log=w_log, + ) + + # MATCH_CATALOGUE (optional, environment-expanded path; empty for none): + # take membership and NUMBER from that external catalogue of the same + # image, before the post-processing keys the epoch HDUs on NUMBER. + match_path = ( + config.getexpanded(module_config_sec, "MATCH_CATALOGUE") + if config.has_option(module_config_sec, "MATCH_CATALOGUE") + else "" ) + if match_path: + mc.match_catalogue( + ss_inst.path_output_file, + match_path, + radius=config.getfloat(module_config_sec, "MATCH_RADIUS"), + min_fraction=config.getfloat( + module_config_sec, "MATCH_MIN_FRACTION" + ), + tolerated_unpaired=config.getint( + module_config_sec, "MATCH_TOLERATED_UNPAIRED" + ), + w_log=w_log, + ) + + # Run sextractor post processing + if config.getboolean(module_config_sec, "MAKE_POST_PROCESS"): + pos_params = config.getlist(module_config_sec, "WORLD_POSITION") + ccd_size = config.getlist(module_config_sec, "CCD_SIZE") + ss.make_post_process( + ss_inst.path_output_file, + f_wcs_path, + pos_params, + ccd_size, + w_log=w_log, + ) + + if work_dir != run_dirs["output"]: + for path in [*ss_inst.check_paths.values(), + ss_inst.path_output_file]: + shutil.move( + path, + os.path.join(run_dirs["output"], os.path.basename(path)), + ) # Return stdout and stderr return stdout, stderr diff --git a/src/shapepipe/pipeline/config.py b/src/shapepipe/pipeline/config.py index 2c294ae65..1f49fe043 100644 --- a/src/shapepipe/pipeline/config.py +++ b/src/shapepipe/pipeline/config.py @@ -97,6 +97,29 @@ def getexpanded(self, section, option, **kwargs): """ return self._get(section, _expandvars_strict, option, **kwargs) + def getexpandedboolean(self, section, option, **kwargs): + """Get Expanded Boolean. + + Expand enviroment variables in the value, then read it as a boolean + the way ``getboolean`` does. + + Parameters + ---------- + section : str + Configuration file section + option : str + Configuration file option + + Returns + ------- + bool + The expanded value as a boolean + + """ + return self._convert_to_boolean( + self.getexpanded(section, option, **kwargs) + ) + def getlist(self, section, option, delimiter=",", **kwargs): """Get List. diff --git a/src/shapepipe/pipeline/file_io.py b/src/shapepipe/pipeline/file_io.py index bf245fbd1..bc51cfd89 100644 --- a/src/shapepipe/pipeline/file_io.py +++ b/src/shapepipe/pipeline/file_io.py @@ -1441,7 +1441,7 @@ def add_cols( col_list = self._cat_data[hdu_no].data.columns + fits.ColDefs( [ - self._make_fits_col(col_name, col_data) + self.fits_column(col_name, col_data) for col_name, col_data in columns.items() ] ) @@ -1462,13 +1462,15 @@ def add_cols( memmap=self.use_memmap, ) - def _make_fits_col(self, col_name, col_data): - """Make FITS Column. + @staticmethod + def fits_column(col_name, col_data): + """FITS Column. Build the ``astropy.io.fits.Column`` that :meth:`add_cols` appends for one array: the FITS type from :meth:`_get_fits_col_type`, a repeat count and ``TDIM`` for multi-dimensional arrays, and a width - set by the longest entry for strings. + set by the longest entry for strings. :meth:`save_as_fits` lays out + the columns of a new table HDU the same way. Parameters ---------- @@ -1483,7 +1485,7 @@ def _make_fits_col(self, col_name, col_data): The column """ - data_type = self._get_fits_col_type(col_data) + data_type = FITSCatalogue._get_fits_col_type(col_data) data_shape = col_data.shape[1:] dim = None mem_size = 1 @@ -1575,7 +1577,8 @@ def _append_col(self, column, hdu_no=None): else: raise BaseCatalogue.catalogueNotOpen(self.fullpath) - def _get_fits_col_type(self, col_data): + @staticmethod + def _get_fits_col_type(col_data): """Get FITS Column Type. Get the FITS data type of a given column. diff --git a/tests/module/test_ngmix.py b/tests/module/test_ngmix.py index 7edc506b3..37b3e58b5 100644 --- a/tests/module/test_ngmix.py +++ b/tests/module/test_ngmix.py @@ -660,7 +660,6 @@ def test_process_counts_flagged_fits_across_batches(tmp_path, monkeypatch, flags ) inst = object.__new__(Ngmix) inst._tile_cat_path = "in-memory-tile" - inst._seg_cat_path = None inst._vignet_cat = SimpleNamespace( gal_vign_cat=galaxies, psf_vign_cat=galaxies, close=lambda: None, ) @@ -759,7 +758,6 @@ def capture(stamp, prior, flux_guess, rng, **kwargs): monkeypatch.setattr(Ngmix, "log_mean_ellipticity", lambda self: None) inst = object.__new__(Ngmix) inst._tile_cat_path = "in-memory-tile" - inst._seg_cat_path = None inst._vignet_cat = SimpleNamespace( gal_vign_cat=galaxies, psf_vign_cat=galaxies, close=lambda: None, ) diff --git a/tests/module/test_ngmix_defect_fill.py b/tests/module/test_ngmix_defect_fill.py index e46bbb000..96de7e44c 100644 --- a/tests/module/test_ngmix_defect_fill.py +++ b/tests/module/test_ngmix_defect_fill.py @@ -48,6 +48,7 @@ split_tile_markers, uberseg_mask, ) +from shapepipe.modules.sextractor_package.sextractor_script import cut_stamps # --- prepare_ngmix_weights: the filled set is the defect set --------------- @@ -161,6 +162,50 @@ def test_uberseg_defect_in_neighbour_region_is_filled(): assert w_out[0, 3] == 0.0 and gal_out[0, 3] == gal[0, 3] +def test_committed_blend_handling_fills_defects(): + """The committed universe's blend_handling is the workflow's default, + and under it every defect is zero-weighted and filled exactly as under + noisefill: the fill does not depend on the blend handling. + + Failure mode: the committed blend handling skips or changes the defect + fill, so the record claims a fill the default campaign does not run + (raw defects, which astra excludes, reaching metacal). + """ + import yaml + + repo = Path(__file__).resolve().parents[2] + universe = yaml.safe_load((repo / "universes" / "committed.yaml").read_text()) + decisions = universe["analyses"]["shape_measurement"]["decisions"] + blend_handling = decisions["blend_handling"] + assert blend_handling in ngmix_module.BLEND_HANDLINGS + workflow = yaml.safe_load((repo / "workflow" / "config.yaml").read_text()) + assert workflow["blend_handling"] == blend_handling + + n = 21 + gal = 1.0e3 + np.random.default_rng(1).normal(0.0, 1.0, (n, n)) + weight = np.ones((n, n)) + flag = np.zeros((n, n), dtype=np.int32) + flag[3, 15] = 1 + flag[10, 2:9] = 2**10 + weight[17, 9] = 0.0 + defect = (weight == 0) | (flag != 0) + kwargs = ( + dict(seg=_uberseg_seg(n), object_number=1) + if blend_handling == "uberseg" + else {} + ) + gal_out, w_out, _ = prepare_ngmix_weights( + gal, weight, flag, np.random.RandomState(0), bkg_rms=np.ones((n, n)), + blend_handling=blend_handling, **kwargs, + ) + ref_out, ref_w, _ = prepare_ngmix_weights( + gal, weight, flag, np.random.RandomState(0), bkg_rms=np.ones((n, n)), + ) + assert np.all(w_out[defect] == 0.0) + assert np.all(gal_out[defect] != gal[defect]), "defects are left raw" + npt.assert_array_equal(gal_out[defect], ref_out[defect]) + + # --- prepare_postage_stamps: the fraction cut counts the defect set -------- N_STAMP = 51 @@ -935,6 +980,140 @@ def stop(*args, **kwargs): assert got is want +# --- Neighbours are markers on another object's segmentation footprint ----- +# +# Under MASK_TYPE CORRECT SExtractor also writes -1e30 over sub-MINAREA sky +# blobs and over fragments CLEAN merged into the object itself. The seg map +# labels those 0 and the object's own NUMBER, so they are not neighbours, and +# the epoch stamps hold real light there: they keep weight and image. + +_OBJ = 1 # the object's NUMBER in _fake_inputs's stores + + +def _seg_marked_tile(): + """Tile VIGNET and SEG_VIGNET with four kinds of interior -1e30 marker. + + Returns ``(tile, seg, kinds)``: a neighbour footprint (label 7), a + detection that left the catalogue (label -1), a fragment of the object + itself (label ``_OBJ``) and a sky blob (label 0); ``kinds`` maps each + name to its marker mask. All of them are off-centre, so the MegaCam flip + moves them. + """ + c = _CENTRE + tile = _tile_with_neighbour() + seg = np.zeros_like(tile, dtype=np.int32) + yy, xx = np.mgrid[:N_STAMP, :N_STAMP] + seg[(yy - c) ** 2 + (xx - c) ** 2 <= 9] = _OBJ + kinds = {name: np.zeros_like(tile, dtype=bool) for name in + ("neighbour", "departed", "own_fragment", "sky")} + kinds["neighbour"][c - 2:c + 3, c + 3:c + 9] = True + kinds["departed"][c - 12:c - 9, c - 12:c - 9] = True + kinds["own_fragment"][c + 6:c + 9, c - 8:c - 5] = True + kinds["sky"][c - 10:c - 8, c + 10] = True + seg[kinds["neighbour"]] = 7 + seg[kinds["departed"]] = -1 + seg[kinds["own_fragment"]] = _OBJ + for mask in kinds.values(): + tile[mask] = _MARKER + return tile, seg, kinds + + +def test_split_tile_markers_keeps_only_other_footprints(): + """With the seg stamp, the neighbour mask is the markers on a label other + than 0 and the object's; the object's own fragments and sky blobs are + dropped, and the off-tile split is unchanged. Without it, every interior + marker is a neighbour. + + Failure mode: noisefill noise-fills the object's own CLEAN-merged light + and sky (neighbours-are-other-footprints). + """ + tile, seg, kinds = _seg_marked_tile() + tile[:, :4] = _MARKER # an off-tile band + neighbour, off_tile = split_tile_markers(tile, tile.shape, seg, _OBJ) + bare_neighbour, bare_off_tile = split_tile_markers(tile, tile.shape) + npt.assert_array_equal(off_tile, bare_off_tile) + assert off_tile[:, :4].all() and not off_tile[:, 4:].any() + npt.assert_array_equal( + neighbour, (kinds["neighbour"] | kinds["departed"]) & ~off_tile + ) + npt.assert_array_equal(bare_neighbour, (tile == _MARKER) & ~off_tile) + + +def _seg_marker_stamp(tile, seg, blend_handling="noisefill"): + clean = np.zeros((N_STAMP, N_STAMP), dtype=np.int32) + ones = np.ones((N_STAMP, N_STAMP)) + epochs = {name: (clean.copy(), ones) for name in _MARKER_EPOCH_NAMES} + vignet, tile_cat, psf_obj, gal_obj = _fake_inputs(epochs) + tile_cat.vign = tile[np.newaxis] + tile_cat.seg = seg[np.newaxis] + return prepare_postage_stamps( + vignet, _OBJ, 0, tile_cat, bkg_sub=False, + psf_obj=psf_obj, gal_obj=gal_obj, blend_handling=blend_handling, + ) + + +def test_noisefill_keeps_light_under_non_neighbour_markers(): + """Through prepare_postage_stamps, each epoch's neighbour mask is its + flipped other-footprint markers; noisefill zero-weights and noise-fills + those and keeps the weight and image of the object's own fragment and + the sky blob. + """ + tile, seg, kinds = _seg_marked_tile() + stamp = _seg_marker_stamp(tile, seg) + assert len(stamp.gals) == len(_MARKER_EPOCH_NAMES) + for i, name in enumerate(stamp.epoch_names): + ccd = int(name.split("-")[1]) + flip = lambda m: Ngmix.MegaCamFlip(m, ccd) + npt.assert_array_equal( + stamp.neighbours[i], + flip(kinds["neighbour"] | kinds["departed"]), + ) + gal, weight, _ = _stamp_epoch_weights(stamp, i, seed=3) + kept = flip(kinds["own_fragment"] | kinds["sky"]) + assert np.all(weight[kept] > 0) + npt.assert_array_equal(gal[kept], 1.0e3 + stamp.gals[i][kept]) + assert np.all(weight[stamp.neighbours[i]] == 0) + + +def test_uberseg_is_unchanged_by_seg_defined_neighbours(monkeypatch): + """Uberseg reads no marker, so defining neighbours by the seg map leaves + every epoch's observation bit-identical: the same epochs survive, with + the same image, weight and noise image. Noisefill does change. + """ + tile, seg, _ = _seg_marked_tile() + marker_split = split_tile_markers + + def observations(blend_handling): + stamp = _seg_marker_stamp(tile, seg, blend_handling) + out = [] + for i in range(len(stamp.gals)): + obs = make_ngmix_observation( + 1.0e3 + stamp.gals[i], stamp.weights[i], stamp.flags[i], + stamp.psfs[i], stamp.jacobs[i], np.random.RandomState(i), + bkg_rms=stamp.bkg_rms[i], offset=stamp.offsets[i], + blend_handling=blend_handling, seg=stamp.segs[i], + object_number=_OBJ, neighbour=stamp.neighbours[i], + ) + out.append((obs.image, obs.weight, obs.noise)) + return stamp.epoch_names, out + + new = {b: observations(b) for b in ("uberseg", "noisefill")} + monkeypatch.setattr( + ngmix_module, "split_tile_markers", + lambda t, shape, seg=None, object_number=None: marker_split(t, shape), + ) + old = {b: observations(b) for b in ("uberseg", "noisefill")} + + assert new["uberseg"][0] == old["uberseg"][0] + for got, want in zip(new["uberseg"][1], old["uberseg"][1]): + for a, b in zip(got, want): + npt.assert_array_equal(a, b) + assert any( + not np.array_equal(got[1], want[1]) + for got, want in zip(new["noisefill"][1], old["noisefill"][1]) + ) + + # --- Off-tile pixels are defects ------------------------------------------- # # The tile VIGNET also holds -1e30 beyond the tile's edge, where the epoch @@ -1054,54 +1233,43 @@ def test_off_tile_pixels_are_zero_weighted_and_filled(blend_handling): npt.assert_array_equal(w_out == 0.0, removed) -# --- Tile VIGNETs cut from a segmentation map ------------------------------ +# --- Tile VIGNETs marked as SExtractor marks them --------------------------- # -# SExtractor's tile VIGNET holds -1e30 off the image and on other objects' -# segmentation footprints. Cutting stamps the same way from a segmentation -# map gives the off-image pixels (off-tile defects) and the footprint pixels -# (the neighbour mask) exactly. +# SExtractor writes -1e30 into the tile VIGNET off the image and on the +# segmentation footprints of other detections. The off-image pixels are the +# off-tile defects and the footprint pixels are the neighbour mask, exactly. DR6_PATCH = Path(__file__).parent / "data" / "dr6_202.301_seg_patch.fits" -def _segmap_stamps(seg, x, y): - """SExtractor-style tile VIGNETs on a unit image, and each stamp's - off-image mask. Each object owns the footprint under its centre pixel.""" - half = N_STAMP // 2 - ny, nx = seg.shape - vignets, off_image = [], [] - for xi, yi in zip(x, y): - r0, c0 = int(np.rint(yi)) - 1, int(np.rint(xi)) - 1 - rows = r0 - half + np.arange(N_STAMP) - cols = c0 - half + np.arange(N_STAMP) - off = ( - ((rows < 0) | (rows >= ny))[:, None] - | ((cols < 0) | (cols >= nx))[None, :] - ) - labels = seg[np.clip(rows, 0, ny - 1)[:, None], - np.clip(cols, 0, nx - 1)[None, :]] - own = seg[r0, c0] - other = (labels != 0) & (labels != own) - vignets.append( - np.where(off | other, _MARKER, 1.0).astype(np.float32) - ) - off_image.append(off) - return np.array(vignets), off_image +def _marked_stamps(seg, x, y): + """Tile VIGNETs on a unit image, marked -1e30 off the image and on every + footprint but the one under the stamp's centre pixel, and each stamp's + off-image mask.""" + col = np.rint(np.asarray(x)).astype(np.int64) - 1 + row = np.rint(np.asarray(y)).astype(np.int64) - 1 + seg_stamps = cut_stamps(seg.astype(np.int64), col, row, N_STAMP, -1) + off_image = [s == -1 for s in seg_stamps] + own = seg[row, col] + vignets = np.ones(seg_stamps.shape, np.float32) + for vign, s, o in zip(vignets, seg_stamps, own): + vign[(s == -1) | ((s != 0) & (s != o))] = _MARKER + return vignets, off_image -def test_dr6_segmap_stamps_split_into_off_image_and_neighbours(): - """On the real 202.301 segmentation patch, every stamp splits into its - off-image pixels (off-tile) and its other -1e30 pixels (neighbours). +def test_dr6_marked_stamps_split_into_off_image_and_neighbours(): + """On the real 202.301 segmentation patch, every marked stamp splits into + its off-image pixels (off-tile) and its other -1e30 pixels (neighbours). - Failure modes: a float32 -1e30 is not recognised as a marker; - off-image pixels of an edge stamp land in the neighbour mask; + Failure modes: a float32 -1e30 is not recognised as a + marker; off-image pixels of an edge stamp land in the neighbour mask; neighbour-footprint pixels become off-tile defects. """ with fits.open(DR6_PATCH) as hdul: seg = hdul["SEG"].data objects = hdul["OBJECTS"].data x, y = np.array(objects["X_IMAGE"]), np.array(objects["Y_IMAGE"]) - vignets, off_image = _segmap_stamps(seg, x, y) + vignets, off_image = _marked_stamps(seg, x, y) assert vignets.dtype == np.float32 n_edge = 0 for vign, off in zip(vignets, off_image): @@ -1130,7 +1298,7 @@ def test_a_neighbour_completing_rows_beside_the_tile_edge_stays_a_neighbour(): seg[20:24, 0:50] = 5 seg[27:32, 19:24] = 1 x, y = np.array([21.0, 30.0]), np.array([30.0, 22.0]) - vignets, off_image = _segmap_stamps(seg, x, y) + vignets, off_image = _marked_stamps(seg, x, y) tile, off = vignets[0], off_image[0] footprint = (tile == _MARKER) & ~off assert footprint.sum() == 4 * (N_STAMP - 5) diff --git a/tests/module/test_ngmix_tile_cat_chunk.py b/tests/module/test_ngmix_tile_cat_chunk.py index 43e58b4cf..49d3ed78c 100644 --- a/tests/module/test_ngmix_tile_cat_chunk.py +++ b/tests/module/test_ngmix_tile_cat_chunk.py @@ -2,8 +2,8 @@ Each ngmix chunk reads the tile catalogue for rows ``chunk_rows(n_obj, row_min, row_max)``. The per-object columns stay full length; the stamp -columns (``VIGNET`` and the segmentation ``VIGNET``) are held for the chunk's -rows only, indexed by tile-catalogue row exactly like the full column. +columns (``VIGNET`` and ``SEG_VIGNET``) are held for the chunk's rows only, +indexed by tile-catalogue row exactly like the full column. """ import numpy as np @@ -31,8 +31,8 @@ def _ldac(path, columns): @pytest.fixture -def catalogues(tmp_path): - """A tile catalogue and its row-aligned segmentation catalogue.""" +def catalogue(tmp_path): + """A tile catalogue with VIGNET and SEG_VIGNET stamp columns.""" rng = np.random.default_rng(3) number = rng.permutation(np.arange(1, 10 * N_OBJ, 10))[:N_OBJ] vign = rng.normal(size=(N_OBJ, STAMP, STAMP)).astype(np.float32) @@ -48,14 +48,11 @@ def catalogues(tmp_path): fits.Column( name="VIGNET", format=f"{STAMP * STAMP}E", dim=dim, array=vign ), - ]) - seg_cat = _ldac(tmp_path / "seg.fits", [ - fits.Column(name="NUMBER", format="J", array=number), fits.Column( - name="VIGNET", format=f"{STAMP * STAMP}J", dim=dim, array=seg + name="SEG_VIGNET", format=f"{STAMP * STAMP}J", dim=dim, array=seg ), ]) - return cat, seg_cat + return cat def _full_columns(path): @@ -69,13 +66,11 @@ def _full_columns(path): "row_min, row_max", [(-1, -1), (1, 6), (7, 15), (16, N_OBJ), (20, -1), (N_OBJ + 1, N_OBJ)], ) -def test_chunk_holds_exactly_its_rows(catalogues, row_min, row_max): - cat, seg_cat = catalogues - full = _full_columns(cat) - full_seg = _full_columns(seg_cat)["VIGNET"] +def test_chunk_holds_exactly_its_rows(catalogue, row_min, row_max): + full = _full_columns(catalogue) rows = chunk_rows(N_OBJ, row_min, row_max) - tile = Tile_cat(cat, seg_cat, row_min=row_min, row_max=row_max) + tile = Tile_cat(catalogue, row_min=row_min, row_max=row_max) assert tile.rows == rows # Per-object columns: full length, every row. @@ -84,7 +79,8 @@ def test_chunk_holds_exactly_its_rows(catalogues, row_min, row_max): np.testing.assert_array_equal(tile.dec, full["YWIN_WORLD"]) np.testing.assert_array_equal(tile.flux, full["FLUX_AUTO"]) - for stamps, column in ((tile.vign, full["VIGNET"]), (tile.seg, full_seg)): + for stamps, column in ((tile.vign, full["VIGNET"]), + (tile.seg, full["SEG_VIGNET"])): # Same stamps, bit for bit, at the same tile-catalogue rows. for i_tile in rows: assert stamps[i_tile].dtype == column[i_tile].dtype @@ -98,15 +94,16 @@ def test_chunk_holds_exactly_its_rows(catalogues, row_min, row_max): stamps[i_tile] -def test_stamps_outlive_the_catalogue_file(catalogues, tmp_path): +def test_stamps_outlive_the_catalogue_file(catalogue): """The held stamps are copies, not views into the memory-mapped file.""" - cat, seg_cat = catalogues - expected = _full_columns(cat)["VIGNET"][4:9].tobytes() - tile = Tile_cat(cat, seg_cat, row_min=5, row_max=9) + full = _full_columns(catalogue) + tile = Tile_cat(catalogue, row_min=5, row_max=9) # Overwrite the file in place with different stamps. - with fits.open(cat, mode="update") as hdul: + with fits.open(catalogue, mode="update") as hdul: hdul[2].data["VIGNET"][:] = 0 - assert tile.vign._stamps.tobytes() == expected + hdul[2].data["SEG_VIGNET"][:] = 0 + assert tile.vign._stamps.tobytes() == full["VIGNET"][4:9].tobytes() + assert tile.seg._stamps.tobytes() == full["SEG_VIGNET"][4:9].tobytes() def test_chunk_stamps_index_by_tile_row(): @@ -117,17 +114,3 @@ def test_chunk_stamps_index_by_tile_row(): for i_tile in (2, 6, -1): with pytest.raises(IndexError): stamps[i_tile] - - -def test_misaligned_seg_catalogue_still_raises(catalogues, tmp_path): - cat, _ = catalogues - short = _ldac(tmp_path / "short_seg.fits", [ - fits.Column(name="NUMBER", format="J", array=np.arange(1, 4)), - fits.Column( - name="VIGNET", format=f"{STAMP * STAMP}J", - dim=f"({STAMP}, {STAMP})", - array=np.zeros((3, STAMP, STAMP), dtype=np.int32), - ), - ]) - with pytest.raises(ValueError, match="row-aligned"): - Tile_cat(cat, short, row_min=1, row_max=5) diff --git a/tests/module/test_ngmix_uberseg.py b/tests/module/test_ngmix_uberseg.py index 183fc046b..4be3ee697 100644 --- a/tests/module/test_ngmix_uberseg.py +++ b/tests/module/test_ngmix_uberseg.py @@ -17,11 +17,13 @@ import numpy as np import numpy.testing as npt import pytest +from astropy.io import fits from scipy import ndimage from sqlitedict import SqliteDict from shapepipe.modules.ngmix_package.ngmix import ( Ngmix, + Tile_cat, central_seg_label, prepare_ngmix_weights, seg_has_neighbour, @@ -330,60 +332,100 @@ def test_check_central_seg_label_missing_number_raises(tmp_path): ngmix._check_central_seg_label(seg, obj_id=99) -# --- FIX 3a: uberseg without seg_cat_path fails at construction ------------- - -def test_ngmix_init_uberseg_without_seg_cat_raises(tmp_path): - """blend_handling='uberseg' with seg_cat_path=None raises at __init__.""" +# --- the seg stamps ride the tile catalogue as SEG_VIGNET ------------------- + +def _write_tile_cat(path, seg_vignets=None): + """A three-object LDAC tile catalogue, with SEG_VIGNET when given.""" + n_obj, size = 3, 5 + cols = [ + fits.Column(name="NUMBER", format="J", array=np.array([4, 5, 9])), + fits.Column(name="XWIN_WORLD", format="D", array=np.zeros(n_obj)), + fits.Column(name="YWIN_WORLD", format="D", array=np.zeros(n_obj)), + fits.Column(name="VIGNET", format=f"{size * size}E", + array=np.ones((n_obj, size * size), np.float32), + dim=f"({size},{size})"), + ] + if seg_vignets is not None: + cols.append(fits.Column(name="SEG_VIGNET", format=f"{size * size}J", + array=seg_vignets.reshape(n_obj, -1), + dim=f"({size},{size})")) + imhead = fits.BinTableHDU.from_columns( + [fits.Column(name="Field Header Card", format="1A", array=["x"])], + name="LDAC_IMHEAD", + ) + fits.HDUList([ + fits.PrimaryHDU(), imhead, + fits.BinTableHDU.from_columns(cols, name="LDAC_OBJECTS"), + ]).writeto(path) + return path + + +def test_tile_cat_reads_seg_vignet(tmp_path): + """Tile_cat's seg stamps are the catalogue's SEG_VIGNET, row for row; + a catalogue without the column has none.""" + seg = np.arange(3 * 25, dtype=np.int32).reshape(3, 5, 5) + with_seg = Tile_cat(str(_write_tile_cat(tmp_path / "a.fits", seg))) + for i_tile in range(len(seg)): + npt.assert_array_equal(with_seg.seg[i_tile], seg[i_tile]) + assert with_seg.seg[i_tile].dtype.kind == "i" + assert Tile_cat(str(_write_tile_cat(tmp_path / "b.fits"))).seg is None + + +def test_uberseg_without_seg_vignet_fails_loudly(tmp_path): + """BLEND_HANDLING = uberseg on a catalogue without SEG_VIGNET raises + before any object is fitted, rather than dropping every object.""" + cat = _write_tile_cat(tmp_path / "tile_cat.fits") names = ("gal", "bkg", "psf", "weight", "flag", "headers") paths = [tmp_path / f"{name}.sqlite" for name in names] for path in paths: SqliteDict(str(path)).close() - with pytest.raises(ValueError, match="requires SEG_VIGNET_PATH"): - Ngmix( - ["tile_cat.fits"] + [str(p) for p in paths[:5]], - str(tmp_path), "-001-001", 30.0, str(paths[5]), - _RecordingLogger(), - blend_handling="uberseg", - seg_cat_path=None, - ) - - -# --- FIX 3b: runner raises when SEG_VIGNET_PATH is set but missing ---------- - -class _FakeConfig: - """Config stub: SEG_VIGNET_PATH is the only present option, and it resolves - to ``seg_path`` (a path the test leaves nonexistent).""" - - def __init__(self, seg_path): - self._seg_path = seg_path - - def getfloat(self, _sec, _key): - return 30.0 - - def getboolean(self, _sec, _key, fallback=False): - return fallback - - def has_option(self, _sec, key): - return key == "SEG_VIGNET_PATH" - - def getexpanded(self, _sec, _key): - return self._seg_path - - -def test_runner_missing_seg_vignet_file_raises(tmp_path): - """SEG_VIGNET_PATH configured but the resolved file is absent -> the runner - fails fast with FileNotFoundError, before constructing Ngmix.""" - from shapepipe.modules.ngmix_runner import ngmix_runner - - input_file_list = ["tile_cat.fits"] + [f"in{i}.sqlite" for i in range(6)] - seg_path = str(tmp_path / "does_not_exist.fits") - with pytest.raises(FileNotFoundError, match="Segmentation vignet file"): - ngmix_runner( - input_file_list, - {"output": str(tmp_path)}, - "-001-001", - _FakeConfig(seg_path), - "NGMIX_RUNNER", - _RecordingLogger(), - ) - + ngmix = Ngmix( + [str(cat)] + [str(p) for p in paths[:5]], + str(tmp_path), "-001-001", 30.0, str(paths[5]), + _RecordingLogger(), blend_handling="uberseg", + ) + with pytest.raises(ValueError, match="SEG_VIGNET"): + ngmix.process() + + +def test_runner_reads_blend_handling_from_the_environment( + tmp_path, monkeypatch, +): + """The committed ini's ``${SP_BLEND_HANDLING:-noisefill}`` reaches Ngmix + as noisefill when the variable is unset and as its value when set.""" + from shapepipe.modules import ngmix_runner as runner_module + from shapepipe.pipeline.config import CustomParser + + names = ("cat", "gal", "bkg", "psf", "weight", "flag", "headers") + paths = [str(tmp_path / f"{name}.sqlite") for name in names] + for path in paths[1:]: + SqliteDict(path).close() + config = CustomParser() + config.add_section("NGMIX_RUNNER") + for key, value in {"MAG_ZP": "30.0", "ID_OBJ_MIN": "-1", + "ID_OBJ_MAX": "-1", + "BLEND_HANDLING": "${SP_BLEND_HANDLING:-noisefill}", + }.items(): + config.set("NGMIX_RUNNER", key, value) + + seen = [] + + class _Stop(Exception): + pass + + def fake_ngmix(*_args, **kwargs): + seen.append(kwargs["blend_handling"]) + raise _Stop + + monkeypatch.setattr(runner_module, "Ngmix", fake_ngmix) + runner = getattr(runner_module.ngmix_runner, "__wrapped__", + runner_module.ngmix_runner) + for env in (None, "uberseg"): + if env is None: + monkeypatch.delenv("SP_BLEND_HANDLING", raising=False) + else: + monkeypatch.setenv("SP_BLEND_HANDLING", env) + with pytest.raises(_Stop): + runner(paths, {"output": str(tmp_path)}, "-001-001", config, + "NGMIX_RUNNER", _RecordingLogger()) + assert seen == ["noisefill", "uberseg"] diff --git a/tests/module/test_pipeline.py b/tests/module/test_pipeline.py index 400ca2ec3..125f24dc6 100644 --- a/tests/module/test_pipeline.py +++ b/tests/module/test_pipeline.py @@ -134,6 +134,46 @@ def test_custom_parser_getexpanded_expands_fallback(monkeypatch): ) +def test_custom_parser_expands_a_default_only_when_the_variable_is_unset( + monkeypatch, +): + """``${VAR:-default}`` takes ``default`` when VAR is unset or empty, and + VAR's value otherwise; a bare unset ``$VAR`` still raises.""" + parser = config.CustomParser() + parser.add_section("S") + parser.set("S", "MODE", "${SP_TEST_MODE:-noisefill}") + parser.set("S", "FLAG", "${SP_TEST_FLAG:-False}") + parser.set("S", "BARE", "$SP_TEST_MODE") + + monkeypatch.delenv("SP_TEST_MODE", raising=False) + monkeypatch.delenv("SP_TEST_FLAG", raising=False) + assert parser.getexpanded("S", "MODE") == "noisefill" + assert parser.getexpandedboolean("S", "FLAG") is False + with pytest.raises(ValueError, match="SP_TEST_MODE"): + parser.getexpanded("S", "BARE") + + monkeypatch.setenv("SP_TEST_MODE", "") + assert parser.getexpanded("S", "MODE") == "noisefill" + + monkeypatch.setenv("SP_TEST_MODE", "uberseg") + monkeypatch.setenv("SP_TEST_FLAG", "True") + assert parser.getexpanded("S", "MODE") == "uberseg" + assert parser.getexpanded("S", "BARE") == "uberseg" + assert parser.getexpandedboolean("S", "FLAG") is True + + +def test_custom_parser_expands_a_variable_value_once(monkeypatch): + """A variable's value is inserted as is, never expanded again.""" + monkeypatch.setenv("SP_TEST_ROOT", "/data/$SP_TEST_OTHER") + monkeypatch.setenv("SP_TEST_OTHER", "expanded-again") + parser = config.CustomParser() + parser.add_section("S") + parser.set("S", "PATH", "${SP_TEST_ROOT:-/x}/a $SP_TEST_ROOT/b") + assert parser.getexpanded("S", "PATH") == ( + "/data/$SP_TEST_OTHER/a /data/$SP_TEST_OTHER/b" + ) + + def test_custom_parser_getlist_honours_custom_delimiter(): parser = config.CustomParser() diff --git a/tests/module/test_sextractor_post_process.py b/tests/module/test_sextractor_post_process.py index 446e1fbf6..868559e36 100644 --- a/tests/module/test_sextractor_post_process.py +++ b/tests/module/test_sextractor_post_process.py @@ -195,3 +195,60 @@ def test_exposure_missing_from_header_log_raises(tmp_path): ["XWIN_WORLD", "YWIN_WORLD"], ["0", str(CCD_NPIX), "0", str(CCD_NPIX)], ) + + +def test_post_process_writes_the_file_catalogue_appends_would(tmp_path): + """make_post_process writes the catalogue once; the bytes are those of + one FITSCatalogue.save_as_fits per EPOCH HDU followed by add_col of + N_EPOCH, the per-step route that rewrote the whole file each time.""" + from shapepipe.pipeline import file_io + + exp_names = ["123456", "654321"] + header_files = [] + for name in exp_names: + npy_path = tmp_path / f"headers-{name}.npy" + _write_exposure_headers(npy_path) + header_files.append([str(npy_path)]) + merge_headers(header_files, str(tmp_path), tile_number="54") + sqlite_path = tmp_path / "log_exp_headers54.sqlite" + + # Objects on CCDs 0 and 2 and one on no CCD. + positions = np.array( + [ + _make_ccd_wcs(ccd)[0].all_pix2world([[50.0, 50.0]], 0)[0] + for ccd in (0, 2) + ] + + [[170.0, 10.0]] + ) + cat_path = tmp_path / "sexcat.fits" + _write_sex_ldac(cat_path, exp_names, positions) + ref_path = tmp_path / "reference.fits" + ref_path.write_bytes(cat_path.read_bytes()) + + sextractor_script.make_post_process( + str(cat_path), + str(sqlite_path), + ["XWIN_WORLD", "YWIN_WORLD"], + ["0", str(CCD_NPIX), "0", str(CCD_NPIX)], + ) + + number = np.arange(1, 4, dtype=">i4") + ccd_n = np.array([0, 2, -1], dtype="int32") + ref = file_io.FITSCatalogue( + str(ref_path), + SEx_catalogue=True, + open_mode=file_io.BaseCatalogue.OpenMode.ReadWrite, + ) + ref.open() + for idx, exp in enumerate(exp_names): + epoch = np.array( + list(zip(number, [exp] * 3, ccd_n)), + dtype=[("NUMBER", number.dtype), ("EXP_NAME", " 0.50001: up + (31.4999, 30), + (20.2, 19), + (20.8, 20), + (0.6, 0), # mx = -0.4 truncates to 0 +]) +def test_vignet_centre_is_sextractors_rule(pos, centre): + assert ss.vignet_centre([pos]).tolist() == [centre] + + +def _cut(array, col, row, fill): + """The STAMP stamp of ``array`` with 0-based (row, col) at its centre.""" + half = STAMP // 2 + padded = np.pad(array, half, constant_values=fill) + return padded[row:row + STAMP, col:col + STAMP] + + +def _scene(tmp_path, flat=False): + """A sexcat whose VIGNETs are cut on the hand-written centres. + + ``flat`` makes the image uniform, so no pixel comparison can tell a tie + object's two candidate centres apart. + """ + rng = np.random.default_rng(3) + image = (np.full((NY, NX), 100, np.float32) if flat + else rng.normal(100, 5, (NY, NX)).astype(np.float32)) + seg = np.zeros((NY, NX), np.int32) + for number, _, _, col, row in OBJECTS: + seg[max(row - 2, 0):row + 3, max(col - 2, 0):col + 3] = number + number = np.array([o[0] for o in OBJECTS]) + x = np.array([o[1] for o in OBJECTS]) + y = np.array([o[2] for o in OBJECTS]) + + vignets = np.array([_cut(image, o[3], o[4], BIG) for o in OBJECTS]) + seg_true = np.array([_cut(seg, o[3], o[4], 0) for o in OBJECTS]) + vignets[(seg_true != 0) & (seg_true != number[:, None, None])] = BIG + + paths = {name: str(tmp_path / f"{name}-001-001.fits") + for name in ("segmentation", "sexcat")} + fits.PrimaryHDU(seg).writeto(paths["segmentation"]) + objects = fits.BinTableHDU.from_columns([ + fits.Column(name="NUMBER", format="J", array=number), + fits.Column(name="X_IMAGE", format="E", array=x.astype(np.float32)), + fits.Column(name="Y_IMAGE", format="E", array=y.astype(np.float32)), + fits.Column(name="VIGNET", format=f"{STAMP * STAMP}E", + array=vignets.reshape(len(number), -1), + dim=f"({STAMP},{STAMP})"), + fits.Column(name="THETA_J2000", format="E", + array=np.zeros(len(number))), + fits.Column(name="X_IMAGE_DBL", format="D", array=x), + fits.Column(name="Y_IMAGE_DBL", format="D", array=y), + ], name="LDAC_OBJECTS") + imhead = fits.BinTableHDU.from_columns( + [fits.Column(name="Field Header Card", format="80A", + array=np.array(["HISTORY x"]))], name="LDAC_IMHEAD") + fits.HDUList([fits.PrimaryHDU(), imhead, objects]).writeto( + paths["sexcat"]) + return paths, seg_true + + +def _add(paths): + ss.add_seg_vignet(paths["sexcat"], paths["segmentation"]) + with fits.open(paths["sexcat"]) as hdul: + return [h.name for h in hdul], hdul["LDAC_OBJECTS"].data.copy() + + +def test_neither_rint_nor_float32_positions_give_the_centres(): + """The premise: np.rint of the double positions misses objects 3 and 4, + and SExtractor's rule on the float32 positions misses object 6.""" + x = np.array([o[1] for o in OBJECTS]) + y = np.array([o[2] for o in OBJECTS]) + col = np.array([o[3] for o in OBJECTS]) + row = np.array([o[4] for o in OBJECTS]) + rint_miss = (np.rint(x - 1) != col) | (np.rint(y - 1) != row) + assert np.flatnonzero(rint_miss).tolist() == [2, 3] + x32 = x.astype(np.float32).astype(np.float64) + y32 = y.astype(np.float32).astype(np.float64) + f32_miss = ((ss.vignet_centre(x32) != col) + | (ss.vignet_centre(y32) != row)) + assert np.flatnonzero(f32_miss).tolist() == [5] + + +@pytest.mark.parametrize("flat", [False, True], ids=["noisy", "flat"]) +def test_seg_vignet_is_registered_with_vignet(tmp_path, flat): + """SEG_VIGNET is the check image on VIGNET's grid for every object, the + half-pixel ties included, even where the image cannot tell the two + candidate centres apart: int32, VIGNET's shape, 0 off the image.""" + paths, seg_true = _scene(tmp_path, flat=flat) + names, data = _add(paths) + assert names == ["PRIMARY", "LDAC_IMHEAD", "LDAC_OBJECTS"] + seg_vignets = data["SEG_VIGNET"] + assert seg_vignets.dtype.kind == "i" and seg_vignets.dtype.itemsize == 4 + assert seg_vignets.shape == data["VIGNET"].shape + npt.assert_array_equal(seg_vignets, seg_true) + centre = STAMP // 2 + npt.assert_array_equal(seg_vignets[:, centre, centre], data["NUMBER"]) + + +def test_seg_vignet_replaces_the_double_positions(tmp_path): + """The written catalogue has the columns SExtractor writes without the + double positions, plus SEG_VIGNET, and every other HDU and column as + they were.""" + paths, _ = _scene(tmp_path) + with fits.open(paths["sexcat"]) as hdul: + before = hdul["LDAC_OBJECTS"].data.copy() + imhead = hdul["LDAC_IMHEAD"].data.copy() + _, after = _add(paths) + kept = [name for name in before.names if name not in DBL] + assert after.names == kept + ["SEG_VIGNET"] + for name in kept: + npt.assert_array_equal(after[name], before[name]) + with fits.open(paths["sexcat"]) as hdul: + npt.assert_array_equal(hdul["LDAC_IMHEAD"].data, imhead) + + +def test_seg_vignet_needs_the_double_positions(tmp_path): + paths, _ = _scene(tmp_path) + with fits.open(paths["sexcat"]) as hdul: + objects = hdul["LDAC_OBJECTS"] + cols = [c for c in objects.columns if c.name not in DBL] + hdus = [hdul[0].copy(), hdul[1].copy(), + fits.BinTableHDU.from_columns(cols, name="LDAC_OBJECTS")] + fits.HDUList(hdus).writeto(paths["sexcat"], overwrite=True) + with pytest.raises(ValueError, match="X_IMAGE_DBL"): + _add(paths) + + +def test_param_file_adds_the_double_positions(tmp_path): + """The runner's parameter file is the configured one plus the two + double-precision positions, appended after its last parameter.""" + dot_param = tmp_path / "default.param" + dot_param.write_text("NUMBER # running number\nX_IMAGE\nVIGNET(51,51)") + out = ss.seg_vignet_param_file(str(dot_param), str(tmp_path / "out.param")) + lines = [line.split("#")[0].strip() + for line in open(out).read().splitlines()] + assert [line for line in lines if line] == [ + "NUMBER", "X_IMAGE", "VIGNET(51,51)"] + DBL + assert dot_param.read_text().count("DBL") == 0 + + +def test_sextractor_caller_names_its_check_images(tmp_path): + """The runner finds the SEGMENTATION check image by type.""" + caller = ss.SExtractorCaller( + [str(tmp_path / "image-001-001.fits"), + str(tmp_path / "weight-001-001.fits")], + str(tmp_path), "-001-001", "d.sex", "d.param", "d.conv", + True, False, False, False, False, False, False, + check_image=["BACKGROUND", "SEGMENTATION"], + ) + assert caller.check_paths == { + "BACKGROUND": f"{tmp_path}/background-001-001.fits", + "SEGMENTATION": f"{tmp_path}/segmentation-001-001.fits", + } + + +# --- SEG_VIGNET through the join to the UNIONS catalogue ------------------- +# +# UberSeg (uberseg_mask, seg_has_neighbour, Ngmix._check_central_seg_label) +# compares each stamp's labels with the row's NUMBER. The check image is +# labelled with SExtractor's NUMBERs; the join replaces NUMBER with the +# catalogue's, and relabels every stamp through the whole SExtractor -> UNIONS +# map, the footprints of rows it drops included (UNMATCHED_LABEL). The scene +# makes old and new numbers collide: each paired row's new NUMBER is another +# object's SExtractor NUMBER, one of them the dropped detection's. + +CROWD_STAMP = 9 +# (SExtractor NUMBER, x, y, UNIONS NUMBER or None for no partner) +CROWD = [ + (1, 10.0, 10.0, 2), + (2, 13.0, 10.0, None), + (3, 10.0, 13.0, 1), + (4, 22.0, 22.0, 3), +] +EXT_ONLY = (7, 5.0, 25.0) +EXT_HEADER = ( + "# 1 NUMBER Running object number\n" + "# 2 X_IMAGE Object position along x [pixel]\n" + "# 3 Y_IMAGE Object position along y [pixel]\n" +) + + +def _crowd_seg(): + seg = np.zeros((30, 30), np.int32) + for number, x, y, _ in CROWD: + col, row = int(x) - 1, int(y) - 1 + seg[row - 1:row + 2, col - 1:col + 2] = number + return seg + + +def _write_crowd(sexcat_path, seg_path): + """SExtractor's outputs for CROWD: the sexcat (with the double positions + seg_vignet_param_file asks for) and the SEGMENTATION check image.""" + number = np.array([o[0] for o in CROWD], np.int32) + x = np.array([o[1] for o in CROWD]) + y = np.array([o[2] for o in CROWD]) + n = len(number) + size = CROWD_STAMP * CROWD_STAMP + objects = fits.BinTableHDU.from_columns([ + fits.Column(name="NUMBER", format="J", array=number), + fits.Column(name="X_IMAGE", format="E", array=x.astype(np.float32)), + fits.Column(name="Y_IMAGE", format="E", array=y.astype(np.float32)), + fits.Column(name="VIGNET", format=f"{size}E", + array=np.ones((n, size), np.float32), + dim=f"({CROWD_STAMP},{CROWD_STAMP})"), + fits.Column(name="X_IMAGE_DBL", format="D", array=x), + fits.Column(name="Y_IMAGE_DBL", format="D", array=y), + ], name="LDAC_OBJECTS") + imhead = fits.BinTableHDU.from_columns( + [fits.Column(name="Field Header Card", format="80A", + array=np.array(["HISTORY x"]))], name="LDAC_IMHEAD") + fits.HDUList([fits.PrimaryHDU(), imhead, objects]).writeto(sexcat_path) + fits.PrimaryHDU(_crowd_seg()).writeto(seg_path) + + +def _write_crowd_external(path): + rows = [(new, x, y) for _, x, y, new in CROWD if new is not None] + rows.append(EXT_ONLY) + path.write_text(EXT_HEADER + "".join( + f"{n:10d} {x:11.4f} {y:11.4f}\n" for n, x, y in rows)) + return str(path) + + +def _old_stamps(): + """The check image's stamps in SExtractor's numbering, row by row.""" + col = np.array([int(o[1]) - 1 for o in CROWD]) + row = np.array([int(o[2]) - 1 for o in CROWD]) + return ss.cut_stamps(_crowd_seg(), col, row, CROWD_STAMP, 0) + + +def _run_runner(tmp_path, monkeypatch, match): + """sextractor_runner as the workflow runs it under blend_handling: + uberseg, with SExtractor replaced by the CROWD outputs.""" + import logging + + from shapepipe.modules import sextractor_runner as runner_module + from shapepipe.pipeline.config import CustomParser + + out = tmp_path / "output" + tmp = tmp_path / "tmp" + out.mkdir() + tmp.mkdir() + num = "-001-001" + + def fake_execute(command_line): + _write_crowd(out / f"sexcat{num}.fits", out / f"segmentation{num}.fits") + return "", "All done" + + monkeypatch.setattr(runner_module, "execute", fake_execute) + dot_param = tmp_path / "default.param" + dot_param.write_text("NUMBER\nX_IMAGE\nY_IMAGE\nVIGNET(9,9)\n") + ext = _write_crowd_external(tmp_path / "CFIS_cat-001-001.cat") + config = CustomParser() + config["SEXTRACTOR_RUNNER"] = { + "EXEC_PATH": "source-extractor", + "DOT_SEX_FILE": "d.sex", "DOT_PARAM_FILE": str(dot_param), + "DOT_CONV_FILE": "d.conv", "WEIGHT_IMAGE": "True", + "FLAG_IMAGE": "False", "PSF_FILE": "False", + "DETECTION_IMAGE": "False", "DETECTION_WEIGHT": "False", + "ZP_FROM_HEADER": "False", "BKG_FROM_HEADER": "False", + "CHECKIMAGE": "BACKGROUND, SEGMENTATION", + "SEG_VIGNET": "${SP_SEG_VIGNET:-False}", + "MATCH_CATALOGUE": "${SP_MATCH_CATALOGUE:-}", + "MATCH_RADIUS": "1.0", "MATCH_MIN_FRACTION": "0.98", + "MATCH_TOLERATED_UNPAIRED": "20", "MAKE_POST_PROCESS": "False", + } + monkeypatch.setenv("SP_SEG_VIGNET", "True") + monkeypatch.setenv("SP_MATCH_CATALOGUE", ext if match else "") + runner_module.sextractor_runner( + [str(tmp_path / f"image{num}.fits"), + str(tmp_path / f"weight{num}.fits")], + {"output": str(out), "tmp": str(tmp)}, num, config, + "SEXTRACTOR_RUNNER", logging.getLogger("test"), + ) + with fits.open(out / f"sexcat{num}.fits") as hdul: + return hdul["LDAC_OBJECTS"].data.copy() + + +def test_runner_seg_vignet_carries_the_final_number(tmp_path, monkeypatch): + """Through the runner, each row's own footprint carries its final + (UNIONS) NUMBER and nothing else does, though every new NUMBER is some + other object's SExtractor NUMBER; the dropped detection's footprint is + UNMATCHED_LABEL. Fails if the join runs before add_seg_vignet, or relabels + only the paired labels.""" + from shapepipe.modules.sextractor_package import match_catalogue as mc + + data = _run_runner(tmp_path, monkeypatch, match=True) + paired = [i for i, o in enumerate(CROWD) if o[3] is not None] + old = _old_stamps()[paired] + old_number = np.array([CROWD[i][0] for i in paired]) + new_number = np.array([CROWD[i][3] for i in paired]) + npt.assert_array_equal(data["NUMBER"], new_number) + seg = data["SEG_VIGNET"] + assert "X_IMAGE_DBL" not in data.names + centre = CROWD_STAMP // 2 + npt.assert_array_equal(seg[:, centre, centre], data["NUMBER"]) + dropped = CROWD[1][0] + for i in range(len(paired)): + npt.assert_array_equal(seg[i] == new_number[i], + old[i] == old_number[i]) + npt.assert_array_equal(seg[i] == mc.UNMATCHED_LABEL, + old[i] == dropped) + npt.assert_array_equal(seg[i] == 0, old[i] == 0) + # Rows 0 and 1 overlap the dropped detection, whose SExtractor label (2) + # is row 0's new NUMBER. + assert (seg[0] == mc.UNMATCHED_LABEL).any() + assert (old[0] == new_number[0]).any() + + +def test_runner_seg_vignet_without_a_join_keeps_sextractor_numbers( + tmp_path, monkeypatch, +): + """Image simulations (empty MATCH_CATALOGUE): every row stays, and the + stamps are the check image's, labelled with SExtractor's NUMBER.""" + data = _run_runner(tmp_path, monkeypatch, match=False) + npt.assert_array_equal(data["NUMBER"], [o[0] for o in CROWD]) + npt.assert_array_equal(data["SEG_VIGNET"], _old_stamps()) + + +def test_uberseg_sees_the_same_neighbours_after_the_join(tmp_path, + monkeypatch): + """UberSeg only asks own-versus-other, so the relabelled stamps give the + neighbour mask and flag the SExtractor-labelled ones give.""" + from shapepipe.modules.ngmix_package.ngmix import ( + seg_has_neighbour, + uberseg_mask, + ) + + data = _run_runner(tmp_path, monkeypatch, match=True) + paired = [i for i, o in enumerate(CROWD) if o[3] is not None] + old = _old_stamps()[paired] + for i, j in enumerate(paired): + new_seg, number = data["SEG_VIGNET"][i], data["NUMBER"][i] + npt.assert_array_equal( + uberseg_mask(new_seg, number), + uberseg_mask(old[i], CROWD[j][0])) + assert (seg_has_neighbour(new_seg, number) + == seg_has_neighbour(old[i], CROWD[j][0])) + assert seg_has_neighbour(data["SEG_VIGNET"][0], data["NUMBER"][0]) + assert not seg_has_neighbour(data["SEG_VIGNET"][2], data["NUMBER"][2]) + + +def test_a_partial_relabel_would_collide(): + """The negative control: relabelling only the paired labels leaves the + dropped detection's SExtractor label 2 in row 0's stamp, equal to row 0's + new NUMBER, so UberSeg would take the neighbour for the object.""" + old = _old_stamps()[0] + partial = old.copy() + for number, _, _, new in CROWD: + if new is not None: + partial[old == number] = new + assert ((partial == 2) & (old != 1)).any() + + +def _join_with_numbers(case, numbers): + """Join CROWD to a catalogue whose three objects carry ``numbers`` + (written as given), and return the joined NUMBER column.""" + from shapepipe.modules.sextractor_package import match_catalogue as mc + + case.mkdir() + _write_crowd(case / "sexcat.fits", case / "seg.fits") + rows = zip(numbers, [10.0, 10.0, 22.0], [10.0, 13.0, 22.0]) + ext = case / "ext.cat" + ext.write_text(EXT_HEADER + "".join( + f"{n:>10} {x:11.4f} {y:11.4f}\n" for n, x, y in rows)) + mc.match_catalogue(str(case / "sexcat.fits"), str(ext)) + with fits.open(case / "sexcat.fits") as hdul: + return hdul["LDAC_OBJECTS"].data["NUMBER"].tolist() + + +@pytest.mark.parametrize("numbers", [ + [5, 5, 6], # repeated + [0, 5, 6], # not positive + [-3, 5, 6], + ["5.5", 7, 6], # not an integer + ["5.0", 7, 6], # read as a float + [2**31, 5, 6], # past int32 + [2**32 - 1, 5, 6], # would narrow to -1, UNMATCHED_LABEL +]) +def test_join_rejects_numbers_that_could_collide(tmp_path, numbers): + """Unique catalogue NUMBERs that the int32 columns hold exactly are what + keep a relabelled stamp's own footprint apart from every other one.""" + with pytest.raises(ValueError, match="NUMBER"): + _join_with_numbers(tmp_path / "case", numbers) + + +def test_join_accepts_the_largest_int32_number(tmp_path): + from shapepipe.modules.sextractor_package import match_catalogue as mc + + assert _join_with_numbers(tmp_path / "case", [1, mc.MAX_NUMBER, 6]) == [ + 1, mc.MAX_NUMBER, 6] + + +def test_seg_vignet_stays_out_of_the_final_catalogue(tmp_path): + """make_cat drops SEG_VIGNET with VIGNET: the stamps are ngmix inputs, + not catalogue columns.""" + from shapepipe.modules.make_cat_package import make_cat + + _write_crowd(tmp_path / "sexcat-001-001.fits", tmp_path / "seg.fits") + ss.add_seg_vignet(str(tmp_path / "sexcat-001-001.fits"), + str(tmp_path / "seg.fits")) + names = list( + make_cat.read_sextractor_data(str(tmp_path / "sexcat-001-001.fits")) + ) + assert "VIGNET" not in names and "SEG_VIGNET" not in names + assert "NUMBER" in names diff --git a/tests/module/test_sextractor_work_dir.py b/tests/module/test_sextractor_work_dir.py new file mode 100644 index 000000000..496050a90 --- /dev/null +++ b/tests/module/test_sextractor_work_dir.py @@ -0,0 +1,111 @@ +"""UNIT TESTS FOR SEXTRACTOR_RUNNER'S WORK_DIR. + +With ``WORK_DIR`` set, SExtractor writes its catalogue and check images in a +fresh directory under it, the SEG_VIGNET cut and the join rewrite the +catalogue there, and the files are moved to the run's output directory once +complete. The published files must be the bytes the in-place run writes, the +work directory must not outlive the run, and a failed run must publish +nothing. + +SExtractor is replaced by the CROWD scene of test_sextractor_seg_vignet, +written to the paths the command line names. +""" + +import logging +import re + +import numpy as np +import pytest +from astropy.io import fits + +from shapepipe.modules import sextractor_runner as runner_module +from shapepipe.pipeline.config import CustomParser +from tests.module.test_sextractor_seg_vignet import ( + EXT_HEADER, + _write_crowd, + _write_crowd_external, +) + +NUM = "-001-001" +FILES = [f"sexcat{NUM}.fits", f"background{NUM}.fits", + f"segmentation{NUM}.fits"] + + +def _run(root, monkeypatch, work_dir=None, ext=None): + """sextractor_runner under blend_handling: uberseg with the DR6 join, in + place or through ``work_dir``; returns the SExtractor command line.""" + out = root / "output" + tmp = root / "tmp" + out.mkdir(parents=True) + tmp.mkdir() + calls = [] + + def fake_execute(command_line): + calls.append(command_line) + cat = re.search(r"-CATALOG_NAME (\S+)", command_line).group(1) + background, seg = re.search( + r"-CHECKIMAGE_NAME (\S+)", command_line).group(1).split(",") + _write_crowd(cat, seg) + fits.PrimaryHDU(np.zeros((30, 30), np.float32)).writeto(background) + return "", "All done" + + monkeypatch.setattr(runner_module, "execute", fake_execute) + dot_param = root / "default.param" + dot_param.write_text("NUMBER\nX_IMAGE\nY_IMAGE\nVIGNET(9,9)\n") + if ext is None: + ext = _write_crowd_external(root / f"CFIS_cat{NUM}.cat") + section = { + "EXEC_PATH": "source-extractor", + "DOT_SEX_FILE": "d.sex", "DOT_PARAM_FILE": str(dot_param), + "DOT_CONV_FILE": "d.conv", "WEIGHT_IMAGE": "True", + "FLAG_IMAGE": "False", "PSF_FILE": "False", + "DETECTION_IMAGE": "False", "DETECTION_WEIGHT": "False", + "ZP_FROM_HEADER": "False", "BKG_FROM_HEADER": "False", + "CHECKIMAGE": "BACKGROUND, SEGMENTATION", + "SEG_VIGNET": "True", "MATCH_CATALOGUE": str(ext), + "MATCH_RADIUS": "1.0", "MATCH_MIN_FRACTION": "0.98", + "MATCH_TOLERATED_UNPAIRED": "20", "MAKE_POST_PROCESS": "False", + } + if work_dir is not None: + section["WORK_DIR"] = str(work_dir) + config = CustomParser() + config["SEXTRACTOR_RUNNER"] = section + runner_module.sextractor_runner( + [str(root / f"image{NUM}.fits"), str(root / f"weight{NUM}.fits")], + {"output": str(out), "tmp": str(tmp)}, NUM, config, + "SEXTRACTOR_RUNNER", logging.getLogger("test"), + ) + return calls[0] + + +def test_work_dir_publishes_the_in_place_bytes(tmp_path, monkeypatch): + """The output directory receives the same files, byte for byte, and + SExtractor wrote them under WORK_DIR, which is left empty.""" + _run(tmp_path / "in_place", monkeypatch) + work = tmp_path / "local" + work.mkdir() + command_line = _run(tmp_path / "staged", monkeypatch, work_dir=work) + + assert f"-CATALOG_NAME {work}/sp-detect{NUM}." in command_line + assert str(tmp_path / "staged" / "output") not in command_line + assert list(work.iterdir()) == [] + staged = tmp_path / "staged" / "output" + assert sorted(p.name for p in staged.iterdir()) == sorted(FILES) + for name in FILES: + assert ((staged / name).read_bytes() + == (tmp_path / "in_place" / "output" / name).read_bytes()) + + +def test_failed_run_publishes_nothing(tmp_path, monkeypatch): + """A join that refuses the catalogue leaves neither a partial sexcat in + the output directory nor the work files on the node.""" + ext = tmp_path / "dup.cat" + ext.write_text(EXT_HEADER + "".join( + f"{n:10d} {x:11.4f} {y:11.4f}\n" + for n, x, y in [(5, 10.0, 10.0), (5, 10.0, 13.0), (6, 22.0, 22.0)])) + work = tmp_path / "local" + work.mkdir() + with pytest.raises(ValueError, match="NUMBER"): + _run(tmp_path / "staged", monkeypatch, work_dir=work, ext=ext) + assert list(work.iterdir()) == [] + assert list((tmp_path / "staged" / "output").iterdir()) == [] diff --git a/tests/unit/test_workflow_tile_detection.py b/tests/unit/test_workflow_tile_detection.py index 498a015de..ae75aac5c 100644 --- a/tests/unit/test_workflow_tile_detection.py +++ b/tests/unit/test_workflow_tile_detection.py @@ -153,3 +153,28 @@ def test_catalogue_retrieve_follows_the_prefix(source, retrieve): config = {"tile_detection": "unions_catalogue", "inputs": {"catalogues": source}} assert run_config.catalogue_source(config) == (source, retrieve) + + +# --- blend handling -------------------------------------------------------- + + +def test_blend_handlings_mirror_the_ngmix_module(): + """run_config.BLEND_HANDLINGS is a copy; the copy must stay true. + + The Snakefile validates `blend_handling:` against it in the launcher venv, + outside the container where shapepipe is importable, so the tuple is + mirrored rather than imported, and read out of the source text here for + the same reason: this file is container-free. + """ + import re + + src = (REPO_ROOT / "src" / "shapepipe" / "modules" / "ngmix_package" + / "ngmix.py").read_text() + match = re.search(r"^BLEND_HANDLINGS = \(([^)]*)\)", src, re.M) + assert match, "ngmix.py no longer defines BLEND_HANDLINGS" + assert _load("run_config").BLEND_HANDLINGS == tuple( + part.strip().strip('"\'') for part in match.group(1).split(",") + if part.strip()) + committed = _load("run_config").load( + str(REPO_ROOT / "workflow" / "config.yaml")) + assert committed["blend_handling"] == "uberseg" diff --git a/tests/workflow/README.md b/tests/workflow/README.md index 509670be3..3e3f2a9e5 100644 --- a/tests/workflow/README.md +++ b/tests/workflow/README.md @@ -44,7 +44,7 @@ The API context remains open while tests inspect jobs and closes before the fixt ## Campaign-boundary pin -`params_pin.json` pins SHA-256 digests of: +`params_pin.json` pins the default campaign (data, psfex, `blend_handling: uberseg`), and `params_pin_noisefill.json` the same campaign with `blend_handling: noisefill`, so a campaign that ran as noisefill stays resumable. Each pins SHA-256 digests of: - `unit_pre()` rendered for every stage; - every rule's shell template; @@ -77,8 +77,11 @@ Apply mutations only to a disposable checkout, run the named test without `--upd | `test_final_cat_merge_reads_every_ready_tile` | Drop one ready tile; append an out-of-scope tile. | | `test_products_use_products_dir_and_run_name` | Rename either merged catalogue or the persist manifest; route products to scratch; derive `CAMPAIGN` from the products directory's basename. | | `test_tile_store_is_unique_per_campaign` | Make `LOCAL_TAG` conditional on `image_sims` (or constant); give one tile_shape member a different store path. | +| `test_blend_handling_reaches_ngmix_only_under_uberseg` | Make the Snakefile or config.yaml default `noisefill`; export `SP_BLEND_HANDLING` under noisefill, or `SP_SEG_VIGNET` only under uberseg; drop `blend_env(...)` from either tile_detect or tile_ngmix; export the wrong value; set `NGMIX_SEG_MEM_MB` or `DETECT_SEG_MEM_MB` to 0; give a committed ini's `SEG_VIGNET` / `BLEND_HANDLING` a literal or the wrong default. | +| `test_unknown_blend_handling_fails_during_parse` | Remove the Snakefile's `blend_handling` check. | | `test_module_logs_outlive_their_run_dirs` | Drop `post=` from tile_vignets or tile_ngmix; point the copy at `$SP_LOCAL` or `$SP_RUN/output`; move it after `exit $rc`. | | `test_missing_run_fails_during_parse` | Remove `run` from `run_config.REQUIRED`; literal paths must still receive the required-key diagnostic, not a later `KeyError`. | | `test_unit_pre_changes_at_campaign_boundary` | Append a line to `unit_pre`; change one rule's `params.pre`; change one shell; change a rendered thread count; make `LOCAL_TAG` empty for data. | +| `test_noisefill_plan_changes_at_campaign_boundary` | Export anything under noisefill; change one rule's `params.pre` or shell. | | `test_params_pin_ignores_fixture_root` | Remove fixture-root or run-dir-hash normalization. | | `test_params_is_a_rerun_trigger_under_both_profiles` | Remove `params` from candide or nibi's `rerun-triggers`. | diff --git a/tests/workflow/conftest.py b/tests/workflow/conftest.py index 134a9c62a..f08beaaa8 100644 --- a/tests/workflow/conftest.py +++ b/tests/workflow/conftest.py @@ -22,7 +22,7 @@ def campaign(request, tmp_path): @pytest.fixture def resolve_dag(monkeypatch): """Expose the resolver so tests can also assert parse-time failures.""" - return lambda campaign: resolve(campaign, monkeypatch) + return lambda campaign, **kwargs: resolve(campaign, monkeypatch, **kwargs) @pytest.fixture diff --git a/tests/workflow/harness.py b/tests/workflow/harness.py index 240c0a7d6..31ba89344 100644 --- a/tests/workflow/harness.py +++ b/tests/workflow/harness.py @@ -206,8 +206,12 @@ def load_profile(name): @contextmanager -def resolve(campaign, monkeypatch): - """Resolve ``all`` in an isolated state directory, without running jobs.""" +def resolve(campaign, monkeypatch, launch_env=None): + """Resolve ``all`` in an isolated state directory, without running jobs. + + ``launch_env`` adds variables to the scrubbed environment, as if the + launching shell had exported them. + """ from snakemake import workflow as sm_workflow scripts = REPO / "workflow" / "scripts" @@ -234,6 +238,8 @@ def resolve(campaign, monkeypatch): "XDG_CACHE_HOME": campaign.root / "cache", }.items(): patch.setenv(name, str(value)) + for name, value in (launch_env or {}).items(): + patch.setenv(name, str(value)) profile = load_profile("candide") try: with SnakemakeApi() as api: diff --git a/tests/workflow/params_pin.json b/tests/workflow/params_pin.json index e6f6615ba..c3d10d39a 100644 --- a/tests/workflow/params_pin.json +++ b/tests/workflow/params_pin.json @@ -13,7 +13,7 @@ "final_cat_merge": "e7f46859c4503a2220713d7bb2507555515d0a9632d780b20f14c59e32210023", "prepare_all_tiles": "b8f872a22adf014e25a7fa5198f49b71a6fe9e56042ed82b682bc8763970a844", "star_cat_merge": "6277450958474af5270982fa35360f2f237a29f7533c526ee9265dfd5acc07a0", - "tile_detect": "1df9ca86e0a22fa65615d4efbcdc90f9bf6f1a7cf71c0a4b9d8ab56b811b6c58", + "tile_detect": "faaf8b4a203229b99d2a8a73931e84b8cce4ab46f281cb74f5d0765c60f1d2b1", "tile_exp_forest": "7447ab4a1049de5f0b5c81e5f9ed2a8644c7bdab85cb0a060bfde81261a89b28", "tile_find_exposures": "c883495a7ec196c61ef770507865f43834b08523793d337e8ddeaef0ad62f117", "tile_get_catalogue": "84ffd03c4bde248c06aa58a0f4b7f399390a0ff3ecbcf07e03d718df4fe0cd7f", @@ -21,12 +21,12 @@ "tile_make_cat": "28e38d428333b7998511e91a2ae5f71beb18da4e8342633a045a92ff706033c2", "tile_merge_cats": "0783c93deeb7f90bdb9772bbc4117851dff1a671518c81f67edd4c7c79cc7e6a", "tile_merge_headers": "3e674f961fd5bb93ce1f3b0e2bb072a25a421f2828315a3db45a960c44a42111", - "tile_ngmix": "a35a6326f345e2fac77444724e917f5a36c3c19377a6c3c66750c8ee32250e62", + "tile_ngmix": "3e6e2fdcd36ccbf42710c9075bce084d890c61d1fcf8f23c610df4711504b37c", "tile_uncompress": "4a013cdd04bc3e3c1082e477161d2b03e0eefe280b1aabb9af25c149a977f55b", "tile_vignets": "1d434b4a6a4bcdeb8f1e5e9a5e4f228b67ade400245b41b033a2c883fb6a3267" }, "schema": 1, - "sha256": "dde1a9f3dbc327a04ec19614df97a6256a4c43967fc2bb3b9d925d2168e3b982", + "sha256": "2ab0ced3a578850e1722a48aea8e434467b9f32a98ad8c6e75a513b4bcb042a3", "unit_pre": { "exp_get_images": "dac6685a207dae3ea81d296ce53e9ba75d4636cf1f81e6399f2dda452680c992", "exp_psf": "87fc8ea1153a709ab7ba56542cfacc041f5f8b7f6cf2f38a9bc9d8004d1cecb4", diff --git a/tests/workflow/params_pin_noisefill.json b/tests/workflow/params_pin_noisefill.json new file mode 100644 index 000000000..09734ef8e --- /dev/null +++ b/tests/workflow/params_pin_noisefill.json @@ -0,0 +1,45 @@ +{ + "algorithm": "sha256", + "rules": { + "all": "572122d8d1901e12ff591b30adf405f8920be8183129befececbb819f3392ed8", + "clean_exposure": "22cb76b13a5205d20a02a9bd3b8c8bea5ea2801555e24dd2b84b11145f7e79d9", + "clean_tile": "c6a91af666e86bd9735effa005dee862e7d3682079eac23c56a877505bdbe76e", + "exp_get_images": "a9e9eca0a99a348e43e0cd44d0c0aae0f890773d03e35007fb77f886b7d21f5f", + "exp_maps": "ee85f159d12abb89ed2423ebf2d6688a59df4167217dec3ecd2cc153d3745b12", + "exp_persist": "302e2837542bc1102430c27c81c600b7cda32e8bddcb5fd60d33950987609fff", + "exp_psf": "87b9fd799a54130eb3dd9c78bb5124838854c23f365df7a991a53f4c6ac4c3ff", + "exp_split": "8fc97dd48e76cd6c7fdf9ee388ba1c1a3dd63428cbe1961641fbebe76b05786e", + "exposure_maps": "da8d3af52e651793a1a158590b802f37cf03bf754e35afce4aab1030082737bf", + "final_cat_merge": "e7f46859c4503a2220713d7bb2507555515d0a9632d780b20f14c59e32210023", + "prepare_all_tiles": "b8f872a22adf014e25a7fa5198f49b71a6fe9e56042ed82b682bc8763970a844", + "star_cat_merge": "6277450958474af5270982fa35360f2f237a29f7533c526ee9265dfd5acc07a0", + "tile_detect": "faaf8b4a203229b99d2a8a73931e84b8cce4ab46f281cb74f5d0765c60f1d2b1", + "tile_exp_forest": "7447ab4a1049de5f0b5c81e5f9ed2a8644c7bdab85cb0a060bfde81261a89b28", + "tile_find_exposures": "c883495a7ec196c61ef770507865f43834b08523793d337e8ddeaef0ad62f117", + "tile_get_catalogue": "84ffd03c4bde248c06aa58a0f4b7f399390a0ff3ecbcf07e03d718df4fe0cd7f", + "tile_get_images": "96e235dad9cdcde8ecf11f4516bc2303dbba1c8115ad343cdb26d3d1513da671", + "tile_make_cat": "28e38d428333b7998511e91a2ae5f71beb18da4e8342633a045a92ff706033c2", + "tile_merge_cats": "0783c93deeb7f90bdb9772bbc4117851dff1a671518c81f67edd4c7c79cc7e6a", + "tile_merge_headers": "3e674f961fd5bb93ce1f3b0e2bb072a25a421f2828315a3db45a960c44a42111", + "tile_ngmix": "a35a6326f345e2fac77444724e917f5a36c3c19377a6c3c66750c8ee32250e62", + "tile_uncompress": "4a013cdd04bc3e3c1082e477161d2b03e0eefe280b1aabb9af25c149a977f55b", + "tile_vignets": "1d434b4a6a4bcdeb8f1e5e9a5e4f228b67ade400245b41b033a2c883fb6a3267" + }, + "schema": 1, + "sha256": "2f502d63b35c11436aa7fe6714794424daa1ed29b914129099eef021533d0639", + "unit_pre": { + "exp_get_images": "dac6685a207dae3ea81d296ce53e9ba75d4636cf1f81e6399f2dda452680c992", + "exp_psf": "87fc8ea1153a709ab7ba56542cfacc041f5f8b7f6cf2f38a9bc9d8004d1cecb4", + "exp_split": "922a62360969a7d8775172debdb77469ab8552d8031fc559d92d63fc30443122", + "tile_detect": "cdd19095ce9e33847284825ed013aa0a35011e9877a3d89d6b8dc0349c6b7707", + "tile_find_exposures": "89f0816fcfb513154aa81783cc361add768b0101b4cce19426bdf4b43db0cf71", + "tile_get_catalogue": "2030933b663be3d229d328ee6266eda5a63c98589cafa1151a203e88c8dc733a", + "tile_get_images": "482acaad46e677ec9d67105c48b5d56116c7dff2d77f4a62bda73061721a7103", + "tile_make_cat": "af03252466961e9613fe800e42dff32ff9a3f476726012a1fe8d1c18067ee95d", + "tile_merge_cats": "9c6cfbe27fa9618df2418953c056c8ad7cdf64d4b6b0d8a73a32168c39b82114", + "tile_merge_headers": "6bd3343d183e5413cedb6b82493445602cc139d4b56fde39281c59085a0efa8d", + "tile_ngmix": "47abe4514dc948fb2115d244853da1f792c9f7eea51a207f64de3ee563dad76c", + "tile_uncompress": "bf181cb759b56143c7f086f614c559d4cececa54ae7f4c55b78fd8bfad096ba5", + "tile_vignets": "348a047b4c2e22d0d68b819a1b0ac3c9677c1ee7914557ac257139290548127a" + } +} diff --git a/tests/workflow/test_dag.py b/tests/workflow/test_dag.py index 89b949855..5077f7ee3 100644 --- a/tests/workflow/test_dag.py +++ b/tests/workflow/test_dag.py @@ -1,5 +1,7 @@ """Resolved-job checks for campaign scope, product paths, and PSF custody.""" +import hashlib +import os import re from collections import Counter from pathlib import Path @@ -245,3 +247,162 @@ def test_mccd_is_refused_during_parse(tmp_path, resolve_dag): with pytest.raises(WorkflowError, match=r"psf_model=mccd: PSF persistence"): with resolve_dag(campaign): pytest.fail("psf_model=mccd must be refused at parse time") + + +# --- blend_handling --------------------------------------------------------- + +# First-attempt memory of the stages that hold SEG_VIGNET, under every +# blend_handling: the base plus the seg stamps' share. +SEG_MEM_MB = {"tile_detect": 4000 + 3000, "tile_ngmix": 5000 + 500} +# The exports every blend_handling sets, and those uberseg adds. +SEG_EXPORTS = {"tile_detect": {"SP_SEG_VIGNET": "True"}} +BLEND_EXPORTS = {"tile_ngmix": {"SP_BLEND_HANDLING": "uberseg"}} +# The option each stage's committed ini reads the export through, and the +# value it must resolve to with and without it. +BLEND_OPTIONS = {"tile_detect": ("SEG_VIGNET", "False", "True"), + "tile_ngmix": ("BLEND_HANDLING", "noisefill", "uberseg")} + + +def _jobs(dag, campaign): + """Every job's prologue, shell and memory, keyed by rule and wildcards, + with the campaign root and the run-dir hash in the node-local store name + made location-free.""" + run_hash = hashlib.sha1(str(campaign.run_dir).encode()).hexdigest()[:8] + + def normalize(value): + if not isinstance(value, str): + return value + return (value.replace(run_hash, "") + .replace(str(campaign.root), "")) + + return { + (job.rule.name, tuple(sorted(job.wildcards_dict.items()))): tuple( + normalize(v) for v in (getattr(job.params, "pre", None), + job.shellcmd, job.resources.get("mem_mb"))) + for job in dag.jobs + } + + +def _surface(campaign, resolve_dag): + """The active rules and :func:`_jobs`.""" + with resolve_dag(campaign) as dag: + return dag.rule_names, _jobs(dag, campaign) + + +def _committed_value(shell, config_dir, option, env, monkeypatch): + """``option`` of the module section of the ini ``shell`` runs, expanded + under ``env`` as ShapePipe expands it.""" + from shapepipe.pipeline.config import CustomParser + + name = shell.split('shapepipe_run -c "$SP_CONFIG/')[1].split('"')[0] + parser = CustomParser() + parser.optionxform = str + assert parser.read(config_dir / name) + section = next(s for s in parser.sections() if s.endswith("_RUNNER")) + for key in ("SP_SEG_VIGNET", "SP_BLEND_HANDLING"): + monkeypatch.delenv(key, raising=False) + for key, value in env.items(): + monkeypatch.setenv(key, value) + return parser.getexpanded(section, option) + + +@pytest.mark.parametrize("detection", ["sextractor", "unions_catalogue"]) +def test_blend_handling_reaches_ngmix_only_under_uberseg( + tmp_path, resolve_dag, monkeypatch, detection): + """A campaign without the knob plans exactly the uberseg campaign. Both + handlings export SP_SEG_VIGNET to tile_detect, which turns its ini's + SEG_VIGNET on, and carry the seg stamps' memory in tile_detect and + tile_ngmix. Against explicit noisefill, uberseg only adds its export to + tile_ngmix's prologue, which turns BLEND_HANDLING to uberseg.""" + surfaces = {} + for blend in (None, "noisefill", "uberseg"): + campaign = Campaign(tmp_path / str(blend), "data", "psfex") + campaign.config["tile_detection"] = detection + if blend is not None: + campaign.config["blend_handling"] = blend + campaign.write_config() + surfaces[blend] = _surface(campaign, resolve_dag) + + assert surfaces["uberseg"] == surfaces[None] + rules, noisefill = surfaces["noisefill"] + uberseg_rules, uberseg = surfaces["uberseg"] + assert uberseg_rules == rules + assert uberseg.keys() == noisefill.keys() + config_dir = Path(__file__).parents[2] / "workflow" / "config" / "cfis" + for key, (pre, shell, mem) in uberseg.items(): + rule = key[0] + nf_pre, nf_shell, nf_mem = noisefill[key] + exports = BLEND_EXPORTS.get(rule, {}) + lines = {f"export {name}='{value}'" for name, value in exports.items()} + + def without_exports(text): + if text is None: + return None + return "\n".join(line for line in text.split("\n") + if line not in lines) + + # The rendered shell carries the prologue, so it moves with it and + # nowhere else. + assert without_exports(shell) == nf_shell, key + if pre is None: + assert nf_pre is None and not exports, key + continue + assert lines <= set(pre.split("\n")), key + assert without_exports(pre) == nf_pre, key + if exports: + option, default, value = BLEND_OPTIONS[rule] + assert _committed_value(shell, config_dir, option, {}, + monkeypatch) == default + assert _committed_value(shell, config_dir, option, exports, + monkeypatch) == value + assert mem == nf_mem, key + if rule in SEG_MEM_MB: + assert mem == SEG_MEM_MB[rule], key + seg_exports = SEG_EXPORTS.get(rule, {}) + if seg_exports: + assert {f"export {name}='{value}'" + for name, value in seg_exports.items()} <= set( + nf_pre.split("\n")), key + option, default, value = BLEND_OPTIONS[rule] + assert _committed_value(shell, config_dir, option, seg_exports, + monkeypatch) == value + + +def test_unknown_blend_handling_fails_during_parse(tmp_path, resolve_dag): + campaign = Campaign(tmp_path / "campaign", "data", "psfex") + campaign.config["blend_handling"] = "mof" + campaign.write_config() + with pytest.raises(WorkflowError, match=r"Invalid blend_handling='mof'"): + with resolve_dag(campaign): + pytest.fail("an unknown blend_handling must fail at parse time") + + +INHERITED_BLEND_ENV = { + f"{prefix}{name}": value + for prefix in ("", "APPTAINERENV_", "SINGULARITYENV_") + for name, value in (("SP_SEG_VIGNET", "True"), + ("SP_BLEND_HANDLING", "uberseg")) +} + + +def test_noisefill_ignores_blend_variables_in_the_launch_shell( + tmp_path, resolve_dag): + """A noisefill campaign launched from a shell that still exports the + uberseg variables plans exactly the campaign launched from a clean one, + and the parse leaves none of them in the environment jobs inherit (the + slurm executor submits with --export=ALL from this process).""" + campaigns = {} + for name in ("clean", "dirty"): + campaigns[name] = Campaign(tmp_path / name, "data", "psfex") + campaigns[name].config["blend_handling"] = "noisefill" + campaigns[name].write_config() + clean, dirty = campaigns["clean"], campaigns["dirty"] + _, clean_jobs = _surface(clean, resolve_dag) + + with resolve_dag(dirty, launch_env=INHERITED_BLEND_ENV) as dag: + leaked = sorted(set(INHERITED_BLEND_ENV) & set(os.environ)) + dirty_jobs = _jobs(dag, dirty) + assert leaked == [] + assert dirty_jobs == clean_jobs + for _, shell, _ in dirty_jobs.values(): + assert "SP_BLEND_HANDLING" not in (shell or "") diff --git a/tests/workflow/test_params_pin.py b/tests/workflow/test_params_pin.py index faeb43158..c97a8177b 100644 --- a/tests/workflow/test_params_pin.py +++ b/tests/workflow/test_params_pin.py @@ -7,6 +7,9 @@ from tests.workflow.params import params_pin PIN = Path(__file__).with_name("params_pin.json") +# blend_handling: noisefill's plan, pinned on its own so a campaign that ran as +# noisefill stays resumable with the knob set explicitly. +NOISEFILL_PIN = Path(__file__).with_name("params_pin_noisefill.json") BOUNDARY_MESSAGE = ( "params.pre is a rerun trigger under both profiles; this change reruns " "every finished unit of a resumed campaign — land it at a campaign " @@ -23,6 +26,21 @@ def test_unit_pre_changes_at_campaign_boundary(psfex_dag, pytestconfig): assert actual == json.loads(PIN.read_text()), BOUNDARY_MESSAGE +def test_noisefill_plan_changes_at_campaign_boundary(tmp_path, resolve_dag, + pytestconfig): + """The same pin for a campaign that sets blend_handling: noisefill.""" + campaign = Campaign(tmp_path / "campaign", "data", "psfex") + campaign.config["blend_handling"] = "noisefill" + campaign.write_config() + with resolve_dag(campaign) as dag: + actual = params_pin(dag) + if pytestconfig.getoption("--update-params-pin"): + NOISEFILL_PIN.write_text( + json.dumps(actual, indent=2, sort_keys=True) + "\n") + assert NOISEFILL_PIN.is_file(), BOUNDARY_MESSAGE + assert actual == json.loads(NOISEFILL_PIN.read_text()), BOUNDARY_MESSAGE + + def test_params_pin_ignores_fixture_root(tmp_path, resolve_dag): """A different temporary campaign directory cannot require a new pin.""" pins = [] diff --git a/universes/committed.yaml b/universes/committed.yaml index 57b413579..79f62f356 100644 --- a/universes/committed.yaml +++ b/universes/committed.yaml @@ -56,7 +56,7 @@ analyses: psf_likelihood_noise: psf_noise_1em5 megacam_ccd_flip: megapipe_flip defect_fill: interpolate - blend_handling: noisefill + blend_handling: uberseg defect_weighting: fourfold_zero epoch_masked_fraction_cut: ten_percent central_defect_veto: fixed_radii diff --git a/workflow/README.md b/workflow/README.md index 5f28c1d31..3ad463b0e 100644 --- a/workflow/README.md +++ b/workflow/README.md @@ -35,6 +35,13 @@ uv pip install 'snakemake>=9,<10' 'snakemake-executor-plugin-slurm>=2.7,<3' # (DR6), or the join fails. `sextractor` (the default for image sims) keeps # SExtractor's own NUMBER. Either way make_cat writes # TILE_UNIQUE_ID = tile_id * 10**6 + NUMBER. +# `blend_handling` is ngmix's neighbour treatment: `uberseg` (the default) or +# `noisefill`. Under both, tile_detect adds the coadd segmentation stamps to +# the sexcat as SEG_VIGNET and ngmix's neighbours are the other objects' +# footprints in them. It is chosen per campaign: flipping it on an existing +# run dir reruns tile_ngmix and the rest of the shape chain for every finished +# tile, and reclaimed exposures make that destructive, so a comparison needs +# its own `run:`. # The committed launcher loads apptainer/1.4.5 + the /project venv, so a # fresh shell always has the right state. diff --git a/workflow/Snakefile b/workflow/Snakefile index 91f6ddc8d..7c7f006fa 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -198,6 +198,25 @@ try: except ValueError as err: raise WorkflowError(str(err)) from None +# How ngmix treats a stamp's neighbours; tile.smk's blend handling section +# carries it to tile_detect and tile_ngmix. +BLEND_HANDLING = config.get("blend_handling", "uberseg") +if BLEND_HANDLING not in run_config.BLEND_HANDLINGS: + raise WorkflowError( + f"Invalid blend_handling={BLEND_HANDLING!r}; expected one of " + f"{sorted(run_config.BLEND_HANDLINGS)}.") + +# The committed inis read SP_SEG_VIGNET and SP_BLEND_HANDLING with the modules' +# own defaults, and only tile.smk's blend_env exports them (SP_BLEND_HANDLING +# under uberseg, the workflow's default). Drop any copy the +# launching shell carries, bare or as the APPTAINERENV_/SINGULARITYENV_ forms +# that survive --cleanenv, so a job inherits none: the slurm executor submits +# with --export=ALL from this process's environment, and each job re-parses +# this file before its shell runs. +for _prefix in ("", "APPTAINERENV_", "SINGULARITYENV_"): + for _name in ("SP_SEG_VIGNET", "SP_BLEND_HANDLING"): + os.environ.pop(_prefix + _name, None) + # SP_PHASE is set by bin/sp and NOWHERE else: `prepare`/`compute` on the two # invocations of `sp run`, `passthrough` on direct commands (`sp --unlock`, `sp # --dag`, `sp exp_psf ...`). It gates the two parse-time side effects — the index diff --git a/workflow/config.yaml b/workflow/config.yaml index 81cb5b4bc..cd577bcd3 100644 --- a/workflow/config.yaml +++ b/workflow/config.yaml @@ -20,6 +20,16 @@ input_type: data # campaign's merged catalogues. Required, and set by the run config. # run: smk-g6 +# How ngmix treats a stamp's neighbours (its BLEND_HANDLING). tile_detect +# writes the coadd segmentation stamps into the sexcat as SEG_VIGNET under +# both, and the neighbours are the other objects' footprints in them: +# `uberseg` zeroes their weight; `noisefill`, ngmix's own default, zero-weights +# and noise-fills the -1e30 VIGNET markers on them. Chosen per campaign: +# flipping it on an existing run dir reruns tile_ngmix and everything after it +# for every finished tile, and reclaimed exposures make that destructive -- a +# comparison needs its own `run:`. +blend_handling: uberseg + # Default settings per input_type, can be overridden by a user-defined run config file. # - tile_detection: both run SExtractor on the tile. `unions_catalogue` (data) # also fetches the UNIONS per-tile catalogue and joins the detections to it, diff --git a/workflow/config/cfis/config_tile_Ng_template.ini b/workflow/config/cfis/config_tile_Ng_template.ini index c02556eea..6e7de81f0 100644 --- a/workflow/config/cfis/config_tile_Ng_template.ini +++ b/workflow/config/cfis/config_tile_Ng_template.ini @@ -139,3 +139,9 @@ MAG_ZP = 30.0 # via ShapePipe's getexpanded (env-expanded, not plain getint). ID_OBJ_MIN = $NGMIX_ROW_MIN ID_OBJ_MAX = $NGMIX_ROW_MAX + +# Neighbour treatment: noisefill, or uberseg, which reads the tile catalogue's +# SEG_VIGNET column. The workflow sets SP_BLEND_HANDLING to uberseg under its +# default blend_handling: uberseg, and leaves it unset under noisefill +# (workflow/config.yaml). +BLEND_HANDLING = ${SP_BLEND_HANDLING:-noisefill} diff --git a/workflow/config/cfis/config_tile_Sx.ini b/workflow/config/cfis/config_tile_Sx.ini index 4bfdf6a24..54d62b8e0 100644 --- a/workflow/config/cfis/config_tile_Sx.ini +++ b/workflow/config/cfis/config_tile_Sx.ini @@ -112,9 +112,24 @@ BKG_FROM_HEADER = False # OBJECTS, -OBJECTS, SEGMENTATION, APERTURES CHECKIMAGE = BACKGROUND, SEGMENTATION +# Write the SEGMENTATION check image's stamps, cut on each VIGNET's grid, as +# the int32 SEG_VIGNET column ngmix's blend handling reads under both +# noisefill and uberseg (needs the SEGMENTATION check image; SExtractor also +# computes X/Y_IMAGE_DBL, which centre the stamps and are left out of the +# sexcat). The workflow sets SP_SEG_VIGNET under every blend_handling +# (workflow/rules/tile.smk). +SEG_VIGNET = ${SP_SEG_VIGNET:-False} + # File name suffix for the output sextractor files (optional) SUFFIX = sexcat +# SExtractor writes its catalogue and check images here, the SEG_VIGNET cut, +# the join and the post-processing rewrite the catalogue here, and the files +# are moved to OUTPUT_DIR once complete. Node-local (/local/scratch, which +# both profiles bind and tile.smk's node-local tile store also uses): each +# whole-catalogue rewrite costs minutes on loaded NFS scratch. +WORK_DIR = /local/scratch + ## Join to an external catalogue (optional) # Path to an ASCII SExtractor catalogue of the same image (NUMBER, X_IMAGE, diff --git a/workflow/rules/tile.smk b/workflow/rules/tile.smk index 8307642f1..8015e690f 100644 --- a/workflow/rules/tile.smk +++ b/workflow/rules/tile.smk @@ -54,6 +54,16 @@ catalogues; unpaired rows leave the catalogue. The catalogue is SExtractor run by MegaPipe on the same DR6 image with the same configuration, so the join is exact; on other pixels (a DR5 image) it fails loudly. +``blend_handling: uberseg``, the default, gives ngmix UberSeg's hard mask +instead of its own noise-fill default. Both read one coadd segmentation stamp +per object from the sexcat's SEG_VIGNET column: uberseg masks by it, and +noisefill takes as neighbours only the -1e30 VIGNET markers on another +object's footprint in it. tile_detect cuts it from SExtractor's SEGMENTATION +check image on the grid of each object's VIGNET, before the join, which +relabels every stamp to the catalogue's NUMBER (-1 on the footprints of rows +that leave, 0 on sky). The switch is one prologue variable and nothing else +(blend_env, below). + There is no `tile_mask` rule, and there will not be one (PR #847). ShapePipe generates no masks: tiles have no instrument flag image of their own, so tile_detect runs SExtractor with FLAG_IMAGE = False against @@ -491,6 +501,25 @@ rule tile_merge_headers: shell: sp_shell("tile_merge_headers", "config_tile_Mh_exp.ini") +# --- blend handling ------------------------------------------------------- + +# The blend-handling exports into a stage's prologue. tile_detect writes +# SEG_VIGNET (config_tile_Sx.ini, SEG_VIGNET = ${SP_SEG_VIGNET:-False}) under +# both handlings: noisefill keeps only the -1e30 markers on another object's +# footprint in it (split_tile_markers), uberseg masks by it. uberseg (the +# default) also sets BLEND_HANDLING (config_tile_Ng_template.ini, +# ${SP_BLEND_HANDLING:-noisefill}), which reruns only tile_ngmix on a flip. +BLEND_ENV = { + "tile_detect": {"SP_SEG_VIGNET": "True"}, + **({"tile_ngmix": {"SP_BLEND_HANDLING": "uberseg"}} + if BLEND_HANDLING == "uberseg" else {}), +} + + +def blend_env(stage): + return BLEND_ENV.get(stage, {}) + + # The UNIONS per-tile catalogue (CFIS..r.cat), from a local mirror or # from vos, the way tile_get_images fetches the image; tile_detect joins its # SExtractor detections to it. Reads only tile_numbers.txt, which unit_pre @@ -528,21 +557,31 @@ if TILE_DETECTION == "unions_catalogue": def detect_env(tile): - """tile_detect's prologue exports: the catalogue to join, if any, and for - image simulations the epoch-membership CCD bounds. + """tile_detect's prologue exports: the catalogue to join, if any, + SP_SEG_VIGNET, and for image simulations the epoch-membership CCD bounds. SP_MATCH_CATALOGUE is empty under tile_detection: sextractor (image simulations), so that no value left exported in the submitting shell reaches the job. SP_CCD_SIZE is exported only for image simulations (their trimmed CCDs); data runs take config_tile_Sx.ini's default. """ - if TILE_DETECTION != "unions_catalogue": - env = {"SP_MATCH_CATALOGUE": ""} - if INPUT_TYPE == "image_sims": - env["SP_CCD_SIZE"] = config["ccd_size"] - return env - gic = f"{tile_dir(tile)}/output/run_sp_tile_Gic/get_images_runner/output" - return {"SP_MATCH_CATALOGUE": f"{gic}/CFIS_cat{unit_num(tile)}.cat"} + env = {"SP_MATCH_CATALOGUE": "", **blend_env("tile_detect")} + if TILE_DETECTION == "unions_catalogue": + gic = f"{tile_dir(tile)}/output/run_sp_tile_Gic/get_images_runner/output" + env["SP_MATCH_CATALOGUE"] = f"{gic}/CFIS_cat{unit_num(tile)}.cat" + elif INPUT_TYPE == "image_sims": + env["SP_CCD_SIZE"] = config["ccd_size"] + return env + + +# tile_detect's extra memory for SEG_VIGNET: add_seg_vignet reads the 400 MB +# SEGMENTATION check image and adds an int32 SEG_VIGNET column the size of +# VIGNET, which the join then holds twice. MEASURED on DR6 tile 202.301 +# (36,022 detections; sextractor_runner as tile_detect runs it, without the +# post-processing): peak RSS 1.53 GiB without SEG_VIGNET, 2.93 GiB with it. +# Scaled to the 1.84 GiB worst tile above, it needs ~3.5 GiB; 7000 MB is 2x +# that. +DETECT_SEG_MEM_MB = 3000 # SExtractor object detection on the tile; under unions_catalogue joined to @@ -573,12 +612,18 @@ rule tile_detect: # size (10000 x 10000 px); SExtractor alone on one takes 34 s and 0.74 GiB. # 4000 MB is 2.1x the worst tile, and an OOM retries at 8000. The runtime # is ~7x the slowest tile, for /scratch I/O on nibi (a 400 MB image and - # weight in, a ~430 MB catalogue out). nibi bills max(cores, mem_GB/4), so - # the job bills 1 core-equivalent. + # weight in, the catalogue and check images out). The SEG_VIGNET cut, the + # join and the post-processing each rewrite the whole catalogue, so + # config_tile_Sx.ini's WORK_DIR keeps them on node-local disk: MEASURED on + # DR6 191.307 under uberseg (31,736 rows, 8 epochs), the step took 44 min + # in the loaded smk-g13-us allocation with the rewrites on /scratch, 112 s + # on idle /scratch, and 36 s through WORK_DIR. nibi bills + # max(cores, mem_GB/4), so the job bills 1 core-equivalent; plus + # DETECT_SEG_MEM_MB above. threads: 1 retries: 1 resources: - mem_mb = lambda wc, attempt: 4000 * attempt, + mem_mb = lambda wc, attempt: (4000 + DETECT_SEG_MEM_MB) * attempt, runtime = lambda wc, attempt: 20 * attempt shell: sp_shell("tile_detect", "config_tile_Sx.ini") @@ -646,6 +691,13 @@ rule tile_vignets: check_args=' --run-dir "$SP_LOCAL" --unit {wildcards.tile}', post=keep_module_logs("$NGMIX_VIGNET_DIR", "run_sp_tile_PiViVi")) +# Each ngmix chunk's extra memory for SEG_VIGNET: Tile_cat holding the tile's +# SEG_VIGNET column, a view into the loaded table like VIGNET. Measured on DR6 +# 202.301 (36065 rows, 51x51): Tile_cat peak and resident RSS 955 MiB with the +# column, 595 without (+360 MiB, the column's size). 500 covers 186.307's +# ~37.9k rows (~375 MiB) with a third to spare. +NGMIX_SEG_MEM_MB = 500 + # ngmix shape measurement — N chunks per tile (D4). Each chunk LOOKS UP its own # CLOSED catalogue-row range in the file tile_vignets materialised at the top of this # group job (TILE_NGMIX_RANGES); the ranges are knowable only at EXECUTION time, @@ -678,7 +730,8 @@ rule tile_ngmix: pre = lambda wc: unit_pre("tile_ngmix", wc.tile, env={"SP_NGMIX_CHUNK": wc.chunk, "NGMIX_N_CHUNKS": NGMIX_CHUNKS, "SP_DEFECT_FILL": DEFECT_FILL, - "SP_DEFECT_WEIGHTING": DEFECT_WEIGHTING}, + "SP_DEFECT_WEIGHTING": DEFECT_WEIGHTING, + **blend_env("tile_ngmix")}, # Two steps, not `eval "$(...)"`: a command substitution inside eval # discards the script's exit status, so a missing sexcat would fall # through to shapepipe_run with an unset range and fail as something @@ -837,7 +890,10 @@ rule tile_ngmix: # margin over the worst tile measured, and the first thing to give would # be the page cache holding the store. Take it only with a measurement # of cache behaviour under pressure, not on the arithmetic alone. - mem_mb = lambda wc, attempt: 5000 * attempt, + # + # Each chunk also holds the tile's SEG_VIGNET column, + # int32 and the size of VIGNET: NGMIX_SEG_MEM_MB below. + mem_mb = lambda wc, attempt: (5000 + NGMIX_SEG_MEM_MB) * attempt, # 120 on the FIRST attempt, and ATTEMPT-SCALED after it. MEASURED # two ways, and the second is why the margin is thinner than it looks: # * alone (job 20795277, tile 198.305 chunk 1): ~76 min for 3547 diff --git a/workflow/scripts/run_config.py b/workflow/scripts/run_config.py index 669a3ec60..7cf9c18c3 100644 --- a/workflow/scripts/run_config.py +++ b/workflow/scripts/run_config.py @@ -28,6 +28,10 @@ # never mention `$run` must still set it. REQUIRED = ("run", "tile_list", "inputs.tiles", "inputs.exposures", "outputs.run_dir", "outputs.index_db") +# ngmix's BLEND_HANDLING values, which `blend_handling` takes. A copy, because +# the Snakefile validates it outside the container; tests/unit holds it equal +# to ngmix_package.ngmix.BLEND_HANDLINGS. +BLEND_HANDLINGS = ("noisefill", "uberseg") def merge(base, over):