From 86641f42e9c289ea1a1d0de617e543e4a73e34ad Mon Sep 17 00:00:00 2001 From: Jean Connelly Date: Thu, 14 May 2026 11:27:55 -0400 Subject: [PATCH 1/7] Use a duration for the standard timing test --- proseco/bright_object.py | 300 +++++++++++++++++++++++++++++ proseco/characteristics_jupiter.py | 27 --- proseco/tests/timing.py | 5 +- 3 files changed, 303 insertions(+), 29 deletions(-) create mode 100644 proseco/bright_object.py delete mode 100644 proseco/characteristics_jupiter.py diff --git a/proseco/bright_object.py b/proseco/bright_object.py new file mode 100644 index 00000000..66559598 --- /dev/null +++ b/proseco/bright_object.py @@ -0,0 +1,300 @@ +from typing import TYPE_CHECKING, NamedTuple + +import astropy.units as u +import numpy as np +from astropy.table import Table +from chandra_aca import planets +from cxotime import CxoTime, CxoTimeLike +from Quaternion import QuatLike + +if TYPE_CHECKING: + from chandra_aca.planets import PlanetPositionTable + + from proseco.core import StarsTable + + +def check_for_close_planets( + date: CxoTimeLike, + duration: float, + att: QuatLike, + tol=2.0, +) -> dict[str, "PlanetPositionTable"]: + from chandra_aca.planets import ( + convert_time_format_spk, + get_planet_angular_sep, + get_planet_chandra_ccd_position, + ) + from cxotime import CxoTime + + date0 = CxoTime(date) + if duration is None: + time_secs = convert_time_format_spk(date0, "secs") + else: + times = date0 + ([0, 0.5, 1] * u.s) * duration + time_secs = convert_time_format_spk(times, "secs") + + planets_dict = {} + for planet in planets.BRIGHT_PLANETS: + sep = get_planet_angular_sep( + planet, + ra=att.ra, + dec=att.dec, + time=time_secs, + observer_position="earth", + ) + if np.all(sep > tol + 0.25): + continue + + sep = get_planet_angular_sep( + planet, + ra=att.ra, + dec=att.dec, + time=time_secs, + observer_position="chandra", + ) + if np.all(sep > tol): + continue + + planets_dict[planet] = get_planet_chandra_ccd_position( + planet=planet, + date=date0, + duration=duration if duration is not None else 0.0, + att=att, + ccd_pad=0.0, + ephem_source="stk", + ) + + return planets_dict + + +def bright_object_distribution_check( + cand_guide_set: Table, + bright_object_data: Table, + dither: float = 4.0, +) -> tuple[bool, bool]: + """ + Check for guide star CCD distribution in presence of a bright object. + + Check that there are at least two candidate guide stars on the side of the CCD that + does not have the bright object. + + Parameters + ---------- + cand_guide_set : Table + Table of candidate guide stars with 'row' column. + bright_object_data : Table + Table with bright object positions with 'row' column. + Returns + ------- + tuple[bool, bool] + Returns ``(distribution_ok, crosses_midline)`` where ``distribution_ok`` + indicates whether the candidate guide stars are correctly distributed with + respect to the bright object and ``crosses_midline`` indicates whether the + bright object padded row range crosses CCD row=0. + """ + # If there is no bright object on CCD, then the check passes. + if len(bright_object_data) == 0: + return True, False + + # It looks like jupiter ang diam goes from 30 to 45 arcsec + # so use 45 / 2 = 22.5 arcsec radius -> 4.5 pixels + # and add a 4 pixel dither pad (default) corresponding to the 20 arcsec HRC pattern + bright_object_size = 4.5 # pixels + sign_max = np.sign(np.max(bright_object_data["row"] + bright_object_size + dither)) + sign_min = np.sign(np.min(bright_object_data["row"] - bright_object_size - dither)) + distribution_ok = ( + np.count_nonzero(np.sign(cand_guide_set["row"]) != sign_max) >= 2 + ) and (np.count_nonzero(np.sign(cand_guide_set["row"]) != sign_min) >= 2) + # Midline crossing is defined at CCD column 0. + sign_max_col = np.sign( + np.max(bright_object_data["col"] + bright_object_size + dither) + ) + sign_min_col = np.sign( + np.min(bright_object_data["col"] - bright_object_size - dither) + ) + crosses_midline = sign_max_col != sign_min_col + + return distribution_ok, crosses_midline + + +def is_spoiled_by_bright_object(cand: Table, bright_object: Table) -> bool: + """ + Check if a single candidate object is spoiled by a bright object. + + This is intended to be used for checking a single fid light, though + could also be used for stars. + + Parameters + ---------- + cand : Table Row + A single astropy Table Row representing the candidate object and + containing 'row' and 'col' columns. + bright_object : Table + Table with bright object positions with 'row' and 'col' columns, or None + if the bright object is not present. + + Returns + ------- + bool + + """ + # convert the cand Table Row into a Table of one row + single_row_table = Table(cand) + return check_spoiled_by_bright_object(single_row_table, bright_object)[0][0] + + +def check_spoiled_by_bright_object( + cands: Table, bright_object: Table, tolerance: int = 15 +) -> tuple[np.ndarray, list[dict]]: + """ + Check which candidates are spoiled by a bright object. + + A candidate is considered spoiled if it is within `tolerance` pixels of the bright + object in column. + + This method also returns a list of rejection info dicts for the spoiled candidates. + + Parameters + ---------- + cands : Table + Table of candidate objects with 'col' columns. + bright_object : Table + Table with bright object positions with 'col' columns. + tolerance : int + The tolerance in pixels for considering a candidate spoiled by the bright object. + Default is 15 pixels. + + Returns + ------- + mask : np.ndarray + A boolean mask on `cands` where True indicates the candidate is spoiled by + the bright object. + rej_info : list of dict + A list of rejection info dicts for the spoiled candidates. + """ + if bright_object is None or len(bright_object) == 0: + return np.zeros(len(cands), dtype=bool), [] + + # Check that the candidates aren't within tolerance columns of the bright object + colmax_obj = np.max(bright_object["col"]) + colmin_obj = np.min(bright_object["col"]) + tol = tolerance # pixels + ok = (cands["col"] < (colmin_obj - tol)) | (cands["col"] > (colmax_obj + tol)) + + # The OK stars are OK the not OK ones are spoiled + if np.all(ok): + return np.zeros(len(cands), dtype=bool), [] + + # Create rejection info dicts + rej_info = [ + { + "id": cands["id"][idx], + "row": cands["row"][idx], + "col": cands["col"][idx], + "reason": "spoiled by bright object", + "stage": 0, + } + for idx in np.where(~ok)[0] + ] + + # return the not-ok mask and the rej_info + return ~ok, rej_info + + +BrightObjectAcqPos = NamedTuple( + "BrightObjectAcqPos", + [ + ("row", float | None), + ("col", float | None), + ], +) + + +def get_bright_object_acq_pos( + date: CxoTimeLike, bright_object: Table +) -> BrightObjectAcqPos: + """ + Get the position of a bright object during acquisition. + + This uses `date` as the acquisition time uses the bright object position at that time. + If the bright object is not on the CCD within 2 ks of acquisition start, returns (None, None). + + Parameters + ---------- + date : CxoTimeLike + The acquisition date. + bright_object : Table + Table with bright object positions with 'time', 'row', and 'col' columns. + + Returns + ------- + acquisition_position : BrightObjectAcqPos + The (row, col) position of the bright object during acquisition, or None, None + if the bright object is not present during acquisition. + """ + # Use 5 minutes as the nominal acquisition time + acq_start = CxoTime(date) + + # If the first time in the bright_object table is not within 2000 seconds then return + # None, None. This reflects a rare but possible situation where the object drifts onto + # the CCD well after the acquisition time. + if ( + len(bright_object) == 0 + or np.abs(bright_object["time"][0] - acq_start.secs) > 2000 + ): + return BrightObjectAcqPos(None, None) + + # Otherwise use the first row and col in the bright_object table + return BrightObjectAcqPos(bright_object["row"][0], bright_object["col"][0]) + + +def add_bright_object_as_acq_spoilers( + date: "CxoTime | CxoTimeLike", + stars: "StarsTable", + bright_object: Table, + mag: float = -3.0, + tolerance: int = 15, +) -> "StarsTable": + """Enforce column keepout zone around bright object using many fake bright stars. + + This adds a bunch of bright objects to the supplied stars table. This is specific to + acquisition as it uses the acquisition time as a the reference time for the position + of the bright object. + + Parameters + ---------- + date : CxoTimeLike + The observation date. + stars : StarsTable + The stars table to which to add the fake stars representing the bright object. + bright_object : Table + Table with bright object positions with 'time', 'row', and 'col' columns. + mag : float, optional + Magnitude of the bright object (default=-3.0 for Jupiter). + tolerance : int, optional + Column tolerance for the keepout zone (default=15 pixels). + + Returns + ------- + StarsTable + A copy of the input `stars` table with the fake stars added. + """ + if len(bright_object) == 0: + return stars + + # Bright object acq position + acq_pos = get_bright_object_acq_pos(date, bright_object=bright_object) + + out = stars.copy() + idincr = 0 + for irow in np.arange(-505, 510, 5): + for icol in np.arange(acq_pos.col - tolerance, acq_pos.col + tolerance + 1, 5): + out.add_fake_star( + row=irow, + col=icol, + mag=mag, + id=20 + idincr, + CLASS=100, + ) + idincr += 1 + return out diff --git a/proseco/characteristics_jupiter.py b/proseco/characteristics_jupiter.py deleted file mode 100644 index 4e992cbb..00000000 --- a/proseco/characteristics_jupiter.py +++ /dev/null @@ -1,27 +0,0 @@ -from astropy.table import Table - - -class JupiterPositionTable(Table): - """ - A subclass of astropy Table for Jupiter positions. - - This is mainly to provide a specific docstring and type alias, but it does - implicitly document that the table should have 'time', 'row', and 'col' columns - and provides an `empty` class method to create an empty table with those columns. - """ - - @classmethod - def empty(cls): - return cls({"time": [], "row": [], "col": []}) - - -# Date range when jupiter vmag >= -2.0 (dim) and doesn't need to be checked. -exclude_dates = [ - {"start": "2024:128:03:25:16.500", "stop": "2024:162:18:00:48.500"}, - {"start": "2025:115:15:28:12.500", "stop": "2025:246:23:43:03.500"}, - {"start": "2026:123:09:50:55.500", "stop": "2026:304:08:56:43.500"}, - {"start": "2027:139:00:21:41.500", "stop": "2027:348:07:56:35.500"}, - {"start": "2028:162:01:00:04.500", "stop": "2029:016:00:45:56.500"}, - {"start": "2029:192:13:47:11.500", "stop": "2030:041:03:26:27.500"}, - {"start": "2030:234:04:44:40.500", "stop": "2031:001:01:00:03.000"}, -] diff --git a/proseco/tests/timing.py b/proseco/tests/timing.py index 4e0722e7..f72e685f 100644 --- a/proseco/tests/timing.py +++ b/proseco/tests/timing.py @@ -23,15 +23,16 @@ def time_get_aca_catalog(n_samples=100): ras = np.random.uniform(0, 360, size=n_samples) decs = np.random.uniform(-90, 90, size=n_samples) rolls = np.random.uniform(0, 360, size=n_samples) + duration = 10000 # Get rid of initial imports or one-time startup stuff (e.g. in AGASC) - get_aca_catalog(**mod_std_info()) + get_aca_catalog(**mod_std_info(duration=duration)) t0 = time.time() for ra, dec, roll in zip(ras, decs, rolls): print(".", end="") sys.stdout.flush() - get_aca_catalog(**mod_std_info(att=(ra, dec, roll))) + get_aca_catalog(**mod_std_info(att=(ra, dec, roll), duration=duration)) t1 = time.time() print() From e641440aed7242396160118a63fc0450950ff6a8 Mon Sep 17 00:00:00 2001 From: Jean Connelly Date: Thu, 14 May 2026 11:28:43 -0400 Subject: [PATCH 2/7] Remove proseco.jupiter --- proseco/jupiter.py | 328 --------------------------------------------- 1 file changed, 328 deletions(-) delete mode 100644 proseco/jupiter.py diff --git a/proseco/jupiter.py b/proseco/jupiter.py deleted file mode 100644 index d51bb8d8..00000000 --- a/proseco/jupiter.py +++ /dev/null @@ -1,328 +0,0 @@ -from typing import TYPE_CHECKING, NamedTuple - -import astropy.units as u -import numpy as np -from astropy.table import Table -from chandra_aca import planets -from chandra_aca.transform import eci_to_radec, radec_to_yagzag, yagzag_to_pixels -from cheta.comps import ephem_stk -from cxotime import CxoTime, CxoTimeLike -from Quaternion import QuatLike - -import proseco.characteristics as ACA -from proseco import characteristics_jupiter -from proseco.characteristics_jupiter import JupiterPositionTable - -if TYPE_CHECKING: - from proseco.core import StarsTable - - -def date_is_excluded(date: CxoTimeLike) -> bool: - """ - Check if the given date is in the list of excluded dates for Jupiter observations. - - If the date is within any of the excluded date ranges, Jupiter checks do not apply, - and this function returns True. - - Parameters - ---------- - date : CxoTimeLike - The date to check. - - Returns - ------- - bool - True if the date is in an excluded range, False otherwise. - """ - date_str = CxoTime(date).date - # Exclude dates are in CXO date format and we compare as strings. This prevents a - # warning from pyerfa about dubius dates in the future where leap seconds may be - # uncertain. - for exclude_date in characteristics_jupiter.exclude_dates: - if exclude_date["start"] <= date_str <= exclude_date["stop"]: - return True - return False - - -def get_jupiter_position( - date: CxoTimeLike, - duration: float, - att: QuatLike, - t_aca: float = ACA.t_aca_default, -) -> JupiterPositionTable: - """ - Get the position of Jupiter on the ACA CCD. - - Parameters - ---------- - date : CxoTimeLike - The start date of the observation (acquisition time) - duration : float - The duration of the observation in seconds. - att : Quaternion or Quat-compatible - The attitude Quaternion. - t_aca : float, optional - The ACA housing temperature in degC to use for the pixel <=> angle transforms. - This defaults to ACA.t_aca_default. - - Returns - ------- - JupiterPositionTable - A table with columns 'time', 'row', 'col' for the times when Jupiter - is on the CCD. If Jupiter is never on the CCD this is a table with zero rows. - """ - date0 = CxoTime(date) - # dates is a 1-d array with a minimum of 2 points (the end points) - dates = CxoTime.linspace(date0, date0 + duration * u.s, step_max=1000 * u.s) - times = dates.secs - - chandra_ephem = ephem_stk.get_ephemeris_stk(start=dates[0], stop=dates[-1]) - - # Get Jupiter positions using chandra_aca.planets - ephem = { - key: np.interp(times, chandra_ephem["time"], chandra_ephem[key]) - for key in ["x", "y", "z"] - } - # shape (len(dates), 3), units km - pos_earth = planets.get_planet_barycentric("earth", dates) - - chandra_eci = np.array( - [ - ephem["x"] / 1000, # convert m from get_ephemeris_stk to km, - ephem["y"] / 1000, - ephem["z"] / 1000, - ] - ).transpose() - eci = planets.get_planet_eci("jupiter", dates, pos_observer=pos_earth + chandra_eci) - - # Convert ECI position to RA, Dec => yag, zag => row, col - ra, dec = eci_to_radec(eci) - yag, zag = radec_to_yagzag(ra, dec, att) - row, col = yagzag_to_pixels(yag, zag, t_aca=t_aca) - - # Row/col limit in pixels to check for bright object - this is padded past the edge of - # the CCD so the checks will continue to run if the coords of Jupiter are off - # the CCD but the extent of Jupiter (~50" diam) could still be imaged on the CCD. - ccd_limit = 512 + 50 / 5 - - # Limit data to entries within or just outside the CCD - ok = (np.abs(row) <= ccd_limit) & (np.abs(col) <= ccd_limit) - out = JupiterPositionTable( - { - "time": times[ok], - "row": row[ok], - "col": col[ok], - } - ) - return out - - -def jupiter_distribution_check( - cand_guide_set: Table, - jupiter_data: JupiterPositionTable, -) -> bool: - """ - Check for guide star CCD distribution in presence of Jupiter. - - Check that there are at least two candidate guide stars on the side of the CCD that - does not have Jupiter. - - Parameters - ---------- - cand_guide_set : Table - Table of candidate guide stars with 'row' column. - jupiter_data : JupiterPositionTable - Table with Jupiter positions with 'row' column. - - Returns - ------- - bool - True if the candidate guide stars are correctly distributed with respect to - Jupiter, False otherwise. - """ - # If there is no Jupiter on CCD, then the check passes. - if len(jupiter_data) == 0: - return True - - # It looks like jupiter ang diam goes from 30 to 45 arcsec - # so use 45 / 2 = 22.5 arcsec radius -> 4.5 pixels - # and add a 4 pixel dither pad corresponding to the 20 arcsec HRC pattern - jupiter_size = 4.5 # pixels - dither = 4 # pixels - sign_max = np.sign(np.max(jupiter_data["row"] + jupiter_size + dither)) - sign_min = np.sign(np.min(jupiter_data["row"] - jupiter_size - dither)) - return (np.count_nonzero(np.sign(cand_guide_set["row"]) != sign_max) >= 2) and ( - np.count_nonzero(np.sign(cand_guide_set["row"]) != sign_min) >= 2 - ) - - -def is_spoiled_by_jupiter(cand: Table, jupiter: JupiterPositionTable) -> bool: - """ - Check if a single candidate object is spoiled by Jupiter. - - This is intended to be used for checking a single fid light, though - could also be used for stars. - - Parameters - ---------- - cand : Table Row - A single astropy Table Row representing the candidate object and - containing 'row' and 'col' columns. - jupiter : JupiterPositionTable - Table with Jupiter positions with 'row' and 'col' columns, or None - if Jupiter is not present. - - Returns - ------- - bool - - """ - # convert the cand Table Row into a Table of one row - single_row_table = Table(cand) - return check_spoiled_by_jupiter(single_row_table, jupiter)[0][0] - - -def check_spoiled_by_jupiter( - cands: Table, jupiter: JupiterPositionTable, tolerance: int = 15 -) -> tuple[np.ndarray, list[dict]]: - """ - Check which candidate are spoiled by Jupiter. - - A candidate is considered spoiled if it is within `tolerance` pixels of Jupiter in - column. - - This method also returns a list of rejection info dicts for the spoiled candidates. - - Parameters - ---------- - cands: Table - Table of candidate objects with 'col' columns. - jupiter: JupiterPositionTable - Table with Jupiter positions with 'col' columns. - tolerance: int - The tolerance in pixels for considering a candidate spoiled by Jupiter. Default - is 15 pixels. - - Returns - ------- - mask: np.ndarray - A boolean mask on `cands` where True indicates the candidate is spoiled by - Jupiter. - rej_info: list of dict - A list of rejection info dicts for the spoiled candidates. - """ - if len(jupiter) == 0: - return np.zeros(len(cands), dtype=bool), [] - - # Check that the candidates aren't within 15 columns of Jupiter - colmax_jup = np.max(jupiter["col"]) - colmin_jup = np.min(jupiter["col"]) - tol = tolerance # pixels - ok = (cands["col"] < (colmin_jup - tol)) | (cands["col"] > (colmax_jup + tol)) - - # The OK stars are OK the not OK ones are spoiled - if np.all(ok): - return np.zeros(len(cands), dtype=bool), [] - - # Create rejection info dicts - rej_info = [ - { - "id": cands["id"][idx], - "row": cands["row"][idx], - "col": cands["col"][idx], - "reason": "spoiled by Jupiter", - "stage": 0, - } - for idx in np.where(~ok)[0] - ] - - # return the not-ok mask and the rej_info - return ~ok, rej_info - - -JupiterAcqPos = NamedTuple( - "JupiterAcqPos", - [ - ("row", float | None), - ("col", float | None), - ], -) - - -def get_jupiter_acq_pos( - date: CxoTimeLike, jupiter: JupiterPositionTable -) -> JupiterAcqPos: - """ - Get the position of Jupiter during acquisition. - - This uses `date` as the acquisition time uses the Jupiter position at that time. - If Jupiter is not on the CCD within 2 ks of acquisition start, returns (None, None). - - Parameters - ---------- - date : CxoTimeLike - The acquisition date. - jupiter : Table - Table with Jupiter positions with 'time', 'row', and 'col' columns. - - Returns - ------- - acquisition_position : JupiterAcqPos - The (row, col) position of Jupiter during acquisition, or None, None - if Jupiter is not present during acquisition. - """ - # Use 5 minutes as the nominal acquisition time - acq_start = CxoTime(date) - - # If the first time in the jupiter table is not within 2000 seconds then return - # None, None. This reflects a rare but possible situation where Jupiter drifts onto - # the CCD well after the acquisition time. - if len(jupiter) == 0 or np.abs(jupiter["time"][0] - acq_start.secs) > 2000: - return JupiterAcqPos(None, None) - - # Otherwise use the first row and col in the jupiter table - return JupiterAcqPos(jupiter["row"][0], jupiter["col"][0]) - - -def add_jupiter_as_lots_of_acq_spoilers( - date: "CxoTime | CxoTimeLike", stars: "StarsTable", jupiter: JupiterPositionTable -) -> "StarsTable": - """Enforce 15-column keepout zone around Jupiter using many fake bright stars. - - This adds a bunch of bright objects to the supplied stars table. This is specific to - acquisition as it uses the acquisition time as a the reference time for the position - of Jupiter. - - Parameters - ---------- - date : CxoTimeLike - The observation date. - stars : StarsTable - The stars table to which to add the fake stars representing Jupiter. - jupiter : JupiterPositionTable - Table with Jupiter positions with 'time', 'row', and 'col' columns. - - Returns - ------- - StarsTable - A copy of the input `stars` table with the fake stars added. - """ - if len(jupiter) == 0: - return stars - - # Jupiter acq position - acq_pos = get_jupiter_acq_pos(date, jupiter=jupiter) - - out = stars.copy() - idincr = 0 - for irow in np.arange(-505, 510, 5): - for icol in np.arange(acq_pos.col - 15, acq_pos.col + 16, 5): - out.add_fake_star( - row=irow, - col=icol, - mag=-3, # V mag of Jupiter - id=20 + idincr, - CLASS=100, - ) - idincr += 1 - return out From a28d6342f78f979900fbd6b1b083b52d75fdb228 Mon Sep 17 00:00:00 2001 From: Jean Connelly Date: Thu, 14 May 2026 11:29:04 -0400 Subject: [PATCH 3/7] Update planet checks --- proseco/acq.py | 29 +++++-- proseco/core.py | 80 +++++++++++------ proseco/fid.py | 7 +- proseco/guide.py | 57 +++++++++---- proseco/tests/test_guide.py | 19 ++++- proseco/tests/test_jupiter.py | 156 ++++++++++++++++++---------------- 6 files changed, 223 insertions(+), 125 deletions(-) diff --git a/proseco/acq.py b/proseco/acq.py index a6f7c319..7a3bb33e 100644 --- a/proseco/acq.py +++ b/proseco/acq.py @@ -255,13 +255,30 @@ def get_acq_catalog(obsid=0, **kwargs): acqs.set_attrs_from_kwargs(obsid=obsid, **kwargs) acqs.set_stars() - # If Jupiter in the target name and field, update the stars with it as a bright - # object. - if len(acqs.jupiter) > 0: - from proseco.jupiter import add_jupiter_as_lots_of_acq_spoilers + # If bright planets are on-CCD, update stars with synthetic spoilers around + # each bright object for acquisition selection. + import astropy.units as u + from chandra_aca.planets import get_planet_mag_states + from cxotime import CxoTime - acqs.stars = add_jupiter_as_lots_of_acq_spoilers( - date=acqs.date, stars=acqs.stars, jupiter=acqs.jupiter + from proseco.bright_object import add_bright_object_as_acq_spoilers + + for planet, planet_pos in acqs.planets.items(): + if len(planet_pos) == 0: + continue + + duration = acqs.duration if acqs.duration is not None else 0 + mag_states = get_planet_mag_states( + planet, start=acqs.date, stop=CxoTime(acqs.date) + duration * u.s + ) + if len(mag_states) == 0: + continue + + acqs.stars = add_bright_object_as_acq_spoilers( + date=acqs.date, + stars=acqs.stars, + bright_object=planet_pos, + mag=np.min(mag_states["mag_start"]), ) # Only allow imposters that are statistical outliers and are brighter than diff --git a/proseco/core.py b/proseco/core.py index c7472f88..02845c9a 100644 --- a/proseco/core.py +++ b/proseco/core.py @@ -38,7 +38,7 @@ get_dark_cal_id = functools.lru_cache(maxsize=6)(get_dark_cal_id) if TYPE_CHECKING: - from proseco.characteristics_jupiter import JupiterPositionTable + from chandra_aca.planets import PlanetPositionTable def to_python(val): @@ -626,38 +626,68 @@ class ACACatalogTable(BaseCatalogTable): log_info = MetaAttribute(default={}, is_kwarg=False) @property - def jupiter(self) -> "JupiterPositionTable": - if hasattr(self, "_jupiter"): - return self._jupiter + def planets(self) -> dict[str, "PlanetPositionTable"]: + """Dictionary of planet positions keyed by planet name (lowercase).""" + if hasattr(self, "_planets"): + return self._planets - from proseco.characteristics_jupiter import JupiterPositionTable + if self.att is None: + return {} - if "jupiter" not in self.target_name.lower(): - self._jupiter = JupiterPositionTable.empty() - return self._jupiter + from proseco.bright_object import check_for_close_planets - from proseco.jupiter import date_is_excluded + self._planets = check_for_close_planets(self.date, self.duration, self.att) - if date_is_excluded(self.date): - self._jupiter = JupiterPositionTable.empty() - return self._jupiter + # Cache brightest state metadata on each planet table so downstream logic + # can use a single source of truth without re-querying state files. + if self.date is not None: + import astropy.units as u + import numpy as np + from chandra_aca.planets import get_planet_mag_states + from cxotime import CxoTime + + duration = self.duration if self.duration is not None else 0.0 + for planet_name, planet_positions in self._planets.items(): + if len(planet_positions) == 0: + continue + + mag_states = get_planet_mag_states( + planet_name, + self.date, + CxoTime(self.date) + duration * u.s, + ) + if len(mag_states) == 0: + continue + + min_state_idx = np.argmin(mag_states["mag_start"]) + action_col = "label" if "label" in mag_states.colnames else "mag_action" + planet_positions.meta["brightest_mag_action"] = str( + mag_states[action_col][min_state_idx] + ) + planet_positions.meta["brightest_mag_start"] = float( + mag_states["mag_start"][min_state_idx] + ) + planet_positions.meta["brightest_mag_stop"] = float( + mag_states["mag_stop"][min_state_idx] + ) - from proseco.jupiter import get_jupiter_position + return self._planets - self._jupiter = get_jupiter_position( - self.date, self.duration, self.att, t_aca=self.t_aca - ) - return self._jupiter + @planets.setter + def planets(self, value: dict[str, Any]) -> None: + """Set planet positions dictionary.""" + from chandra_aca.planets import PlanetPositionTable - @jupiter.setter - def jupiter(self, value: Any) -> None: - from proseco.characteristics_jupiter import JupiterPositionTable + self._planets = {} + if value is None: + return - self._jupiter = ( - value - if isinstance(value, JupiterPositionTable) - else JupiterPositionTable(value) - ) + for planet_name, planet_data in value.items(): + self._planets[planet_name] = ( + planet_data + if isinstance(planet_data, PlanetPositionTable) + else PlanetPositionTable(planet_data) + ) @property def dark(self): diff --git a/proseco/fid.py b/proseco/fid.py index 5fe09c41..215993b3 100644 --- a/proseco/fid.py +++ b/proseco/fid.py @@ -16,8 +16,8 @@ from . import characteristics_acq as ACQ from . import characteristics_fid as FID from . import guide +from .bright_object import is_spoiled_by_bright_object from .core import ACACatalogTable, AliasAttribute, MetaAttribute -from .jupiter import is_spoiled_by_jupiter def get_fid_catalog(obsid=0, **kwargs): @@ -392,7 +392,10 @@ def get_fid_candidates(self): or self.near_hot_or_bad_pixel(fid) or self.has_column_spoiler(fid, self.stars, stars_mask) or self.is_excluded(fid) - or is_spoiled_by_jupiter(fid, self.jupiter) + or any( + is_spoiled_by_bright_object(fid, planet_positions) + for planet_positions in self.planets.values() + ) ) included = fid["id"] in self.include_ids_fid if not included and excluded: diff --git a/proseco/guide.py b/proseco/guide.py index eb2a20e2..48fb107b 100644 --- a/proseco/guide.py +++ b/proseco/guide.py @@ -648,9 +648,33 @@ def index_combinations(n, m): best_score = -1 best_cands = None - # The best possible score is 3 for 3 cluster checks or 8 for 3 cluster checks plus - # a weighted-5 jupiter check - best_possible_score = 8 if self.jupiter else 3 + from proseco.bright_object import bright_object_distribution_check + + # Weight the bright-object distribution check for planets on CCD that are in + # partial or full mitigation states during the observation window. + planets_for_distribution_check = [] + for _, planet_positions in self.planets.items(): + if len(planet_positions) == 0: + continue + + brightest_mag_action = planet_positions.meta.get("brightest_mag_action") + + # Allow tests that directly set _planets and omit metadata. + if brightest_mag_action is None: + planets_for_distribution_check.append(planet_positions) + continue + + if brightest_mag_action not in ("partial mitigation", "full mitigation"): + continue + + planets_for_distribution_check.append(planet_positions) + + planet_check_weight = 5 + # The best possible score is 3 for the 3 cluster checks plus any weighted + # bright-object distribution checks. + best_possible_score = 3 + planet_check_weight * len( + planets_for_distribution_check + ) for comb in index_combinations(len(stage_cands), choose_m): cands = stage_cands[list(comb)] @@ -668,12 +692,11 @@ def index_combinations(n, m): cluster_check_status = run_cluster_checks(cands) score += np.sum(cluster_check_status) * cluster_weights - if len(self.jupiter) > 0: - jupiter_weight = 5 - from proseco.jupiter import jupiter_distribution_check - - jupiter_check_status = jupiter_distribution_check(cands, self.jupiter) - score += np.sum([jupiter_check_status]) * jupiter_weight + for planet_positions in planets_for_distribution_check: + distribution_check_status, _ = bright_object_distribution_check( + cands, planet_positions + ) + score += int(distribution_check_status) * planet_check_weight if score > best_score: best_score = score @@ -1148,18 +1171,18 @@ def get_initial_guide_candidates(self): f"{len(cand_guides)} candidate guide stars" ) - if len(self.jupiter) > 0: - from proseco.jupiter import check_spoiled_by_jupiter + # Filter stars that are spoiled by any planet + from proseco.bright_object import check_spoiled_by_bright_object - # Exclude candidates within 15 columns of Jupiter - spoiled_by_jupiter, jupiter_rej = check_spoiled_by_jupiter( - cand_guides, self.jupiter + for _, planet_positions in self.planets.items(): + spoiled_by_planet, planet_rej = check_spoiled_by_bright_object( + cand_guides, planet_positions ) - if np.any(spoiled_by_jupiter): - for rej in jupiter_rej: + if np.any(spoiled_by_planet): + for rej in planet_rej: rej["stage"] = 0 self.reject(rej) - cand_guides = cand_guides[~spoiled_by_jupiter] + cand_guides = cand_guides[~spoiled_by_planet] # Get the brightest 2x2 in the dark map for each candidate and save value and location imp_mag, imp_row, imp_col = get_imposter_mags( diff --git a/proseco/tests/test_guide.py b/proseco/tests/test_guide.py index e5be1079..d504d24d 100644 --- a/proseco/tests/test_guide.py +++ b/proseco/tests/test_guide.py @@ -8,6 +8,7 @@ import pytest from astropy.table import Table from chandra_aca.aca_image import ACAImage, AcaPsfLibrary +from chandra_aca.planets import PlanetPositionTable from chandra_aca.transform import count_rate_to_mag, mag_to_count_rate from Quaternion import Quat @@ -654,13 +655,27 @@ def test_select_catalog_jupiter_weighted(): assert set(selected1["id"]) == set([2, 3]) # Simulate Jupiter data so only stars on opposite sides pass - # For testing this accesses the internal _jupiter attribute + # For testing this accesses the internal _planets attribute # directly, but this is not part of the public API. - guides._jupiter = Table([{"row": [100], "col": [0]}]) + guides._planets = {"jupiter": Table([{"row": [100], "col": [0]}])} selected2 = guides.select_catalog(stars) # Should select the combo that passes the Jupiter check assert set(selected2["id"]) == set([1, 3]) + # No-action state should not apply the weighted distribution check. + no_action_planet = PlanetPositionTable({"row": [100], "col": [0]}) + no_action_planet.meta["brightest_mag_action"] = "no action" + guides._planets = {"jupiter": no_action_planet} + selected3 = guides.select_catalog(stars) + assert set(selected3["id"]) == set([2, 3]) + + # Partial mitigation state should apply the weighted distribution check. + partial_planet = PlanetPositionTable({"row": [100], "col": [0]}) + partial_planet.meta["brightest_mag_action"] = "partial mitigation" + guides._planets = {"jupiter": partial_planet} + selected4 = guides.select_catalog(stars) + assert set(selected4["id"]) == set([1, 3]) + def test_select_catalog_fallback(): # No combination passes cluster or Jupiter check diff --git a/proseco/tests/test_jupiter.py b/proseco/tests/test_jupiter.py index bebb96f1..2e421c9a 100644 --- a/proseco/tests/test_jupiter.py +++ b/proseco/tests/test_jupiter.py @@ -1,13 +1,12 @@ import numpy as np import pytest from astropy.table import Table -from chandra_aca import planets, transform +from chandra_aca.planets import PlanetPositionTable from cheta import fetch from cxotime import CxoTime from Quaternion import Quat -from proseco import get_aca_catalog, jupiter -from proseco.characteristics_jupiter import JupiterPositionTable +from proseco import bright_object, get_aca_catalog from proseco.core import StarsTable from proseco.tests.test_common import DARK40, mod_std_info @@ -20,51 +19,6 @@ HAS_CHETA_EPHEM = False -@pytest.mark.skipif(not HAS_CHETA_EPHEM, reason="Requires cheta ephemeris access") -def test_jupiter_position(): - """ - Test jupiter.get_jupiter_position - - Test jupiter.get_jupiter_position against chandra_aca.planets.get_planet_chandra for - a known date and attitude. This is from obsid 23375. - - The proseco code is using the stk ephemeris instead of the cheta predictive - ephemeris used by chandra_aca.planets.get_planet_chandra. - """ - att = Quat(q=[-0.51186291, 0.27607314, -0.17243277, 0.79501379]) - date = "2021:290:11:33:16.000" - duration = 36000 - jupiter_proseco_data = jupiter.get_jupiter_position(date, duration, att, t_aca=25.0) - eci = planets.get_planet_chandra("jupiter", jupiter_proseco_data["time"]) - ra, dec = transform.eci_to_radec(eci) - yag, zag = transform.radec_to_yagzag(ra, dec, att) - row, col = transform.yagzag_to_pixels(yag, zag, t_aca=25.0) - - jupiter_aca_data = JupiterPositionTable( - {"time": jupiter_proseco_data["time"], "row": row, "col": col} - ) - # Compare the two tables - assert len(jupiter_proseco_data) == len(jupiter_aca_data) - assert np.allclose( - jupiter_proseco_data["row"], jupiter_aca_data["row"], atol=0.1, rtol=0 - ) - assert np.allclose( - jupiter_proseco_data["col"], jupiter_aca_data["col"], atol=0.1, rtol=0 - ) - - -def test_jupiter_exclude_dates(): - # Dates within the exclude range should return True - assert jupiter.date_is_excluded("2026:150") - assert jupiter.date_is_excluded("2026-05-10") - assert jupiter.date_is_excluded("2026:300") - # Dates outside the exclude range should return False - assert not jupiter.date_is_excluded("2026:100") - assert not jupiter.date_is_excluded("2026:310") - assert not jupiter.date_is_excluded("2025-09-04") - assert not jupiter.date_is_excluded("2027:130") - - def test_jupiter_offset_left(): """ Test a case where Jupiter is on the left side of the CCD. @@ -121,7 +75,7 @@ def test_jupiter_offset_left(): assert len(aca.guides) == 2 # Confirm Jupiter is all on the left side of the CCD - assert np.all(aca.jupiter["row"] < 0) + assert np.all(aca.planets["jupiter"]["row"] < 0) # Confirm two guide stars on the opposite side of the CCD # This checks optimization because there is a brighter star @@ -129,7 +83,7 @@ def test_jupiter_offset_left(): assert np.sum(aca.guides["row"] > 0) >= 2 # Confirm no guide stars within 15 columns of Jupiter - for jcol in aca.jupiter["col"]: + for jcol in aca.planets["jupiter"]["col"]: dcol = np.abs(aca.guides["col"] - jcol) assert np.all(dcol > 15) @@ -171,7 +125,7 @@ def test_jupiter_offset_right(): ) ) # >>> aca.jupiter - # + # # time row col # float64 float64 float64 # ----------------- ------------------ ------------------- @@ -211,7 +165,7 @@ def test_jupiter_offset_right(): assert len(aca.guides) == 2 # Confirm Jupiter is all on the right side of the CCD - assert np.all(aca.jupiter["row"] > 0) + assert np.all(aca.planets["jupiter"]["row"] > 0) # Confirm two guide stars on the opposite side of the CCD # This checks optimization because there is a brighter star @@ -219,7 +173,7 @@ def test_jupiter_offset_right(): assert np.sum(aca.guides["row"] < 0) >= 2 # Confirm no guide stars within 15 columns of Jupiter - for jcol in aca.jupiter["col"]: + for jcol in aca.planets["jupiter"]["col"]: dcol = np.abs(aca.guides["col"] - jcol) assert np.all(dcol > 15) @@ -254,8 +208,8 @@ def test_jupiter_midline(): ) ) # Confirm Jupiter is on both sides of the CCD - assert np.max(aca.jupiter["row"]) > 0 - assert np.min(aca.jupiter["row"]) < 0 + assert np.max(aca.planets["jupiter"]["row"]) > 0 + assert np.min(aca.planets["jupiter"]["row"]) < 0 # Confirm two guide stars on each side of the CCD assert np.sum(aca.guides["row"] > 0) >= 2 @@ -266,7 +220,7 @@ def test_jupiter_midline(): assert 204 in aca.guides["id"] # Confirm no guide stars within 15 columns of Jupiter - for jcol in aca.jupiter["col"]: + for jcol in aca.planets["jupiter"]["col"]: dcol = np.abs(aca.guides["col"] - jcol) assert np.all(dcol > 15) @@ -286,8 +240,12 @@ def test_jupiter_acquisition(col_dist_arcsec): att = Quat(q=[-0.49963289, 0.25613709, -0.16664083, 0.81055018]) stars.att = att date = "2021:249:12:00:00.000" - jupiter_pos = jupiter.get_jupiter_position(date, 30000, att) - jupiter_acq_pos = jupiter.get_jupiter_acq_pos(date, jupiter=jupiter_pos) + from chandra_aca.planets import get_planet_chandra_ccd_position + + jupiter_pos = get_planet_chandra_ccd_position("jupiter", date, 30000, att) + jupiter_acq_pos = bright_object.get_bright_object_acq_pos( + date, bright_object=jupiter_pos + ) col_dist = int(col_dist_arcsec / 5) stars.add_fake_star(id=200, mag=6.5, row=-300, col=400) @@ -390,10 +348,12 @@ def test_get_jupiter_position_returns_table(): date = "2025:093:12:26:04.000" duration = 10500 # seconds att = Quat(q=[-0.43419701, -0.51408310, 0.33920339, 0.65736792]) - out = jupiter.get_jupiter_position(date, duration, att) + from chandra_aca.planets import get_planet_chandra_ccd_position + + out = get_planet_chandra_ccd_position("jupiter", date, duration, att) - # Should return a JupiterPositionTable - assert isinstance(out, JupiterPositionTable) + # Should return a PlanetPositionTable + assert isinstance(out, PlanetPositionTable) assert out.colnames == ["time", "row", "col"] assert len(out) == 12 @@ -404,31 +364,47 @@ def test_get_jupiter_position_returns_table(): def test_jupiter_distribution_check_1(): # Simulate jupiter_data crosses 0 - jupiter_data = JupiterPositionTable({"row": [-10, 20]}) + jupiter_data = PlanetPositionTable({"row": [-10, 20], "col": [-10, 10]}) cand_guide_set = Table({"row": [-400, -300, 400, 300]}) # Should pass: at least two on each side - assert jupiter.jupiter_distribution_check(cand_guide_set, jupiter_data) + distribution_ok, crosses_midline = bright_object.bright_object_distribution_check( + cand_guide_set, jupiter_data + ) + assert distribution_ok + assert crosses_midline # Should fail: only with all on one side cand_guide_set = Table({"row": [-400, -300, -200]}) - assert not jupiter.jupiter_distribution_check(cand_guide_set, jupiter_data) + distribution_ok, crosses_midline = bright_object.bright_object_distribution_check( + cand_guide_set, jupiter_data + ) + assert not distribution_ok + assert crosses_midline def test_jupiter_distribution_check_2(): # Simulate jupiter_data all positive - jupiter_data = JupiterPositionTable({"row": [10, 20]}) + jupiter_data = PlanetPositionTable({"row": [10, 20], "col": [10, 20]}) cand_guide_set = Table({"row": [-400, -300, 400, 300]}) # Should pass: at least two on each side - assert jupiter.jupiter_distribution_check(cand_guide_set, jupiter_data) + distribution_ok, crosses_midline = bright_object.bright_object_distribution_check( + cand_guide_set, jupiter_data + ) + assert distribution_ok + assert not crosses_midline # Should fail: only one on opposite side cand_guide_set = Table({"row": [400, 300, -200]}) - assert not jupiter.jupiter_distribution_check(cand_guide_set, jupiter_data) + distribution_ok, crosses_midline = bright_object.bright_object_distribution_check( + cand_guide_set, jupiter_data + ) + assert not distribution_ok + assert not crosses_midline def test_check_spoiled_by_jupiter(): # Simulate candidate stars and jupiter data cand_stars = Table({"id": [1, 2, 3], "col": [10, 50, 100], "row": [0, 0, 0]}) - jupiter_data = JupiterPositionTable({"col": [40, 60], "row": [0, 0]}) - mask, rej = jupiter.check_spoiled_by_jupiter(cand_stars, jupiter_data) + jupiter_data = PlanetPositionTable({"col": [40, 60], "row": [0, 0]}) + mask, rej = bright_object.check_spoiled_by_bright_object(cand_stars, jupiter_data) # Only star with col=50 should be spoiled (within 15 pixels of Jupiter) assert np.array_equal(mask, [False, True, False]) assert len(rej) == 1 @@ -436,7 +412,7 @@ def test_check_spoiled_by_jupiter(): "id": 2, "row": 0, "col": 50, - "reason": "spoiled by Jupiter", + "reason": "spoiled by bright object", "stage": 0, } @@ -445,7 +421,7 @@ def test_add_jupiter_as_spoilers(): date = "2025:220:12:00:00" stars = StarsTable.empty() jupiter_data = Table({"time": [CxoTime(date).secs], "row": [100], "col": [100]}) - out = jupiter.add_jupiter_as_lots_of_acq_spoilers(date, stars, jupiter_data) + out = bright_object.add_bright_object_as_acq_spoilers(date, stars, jupiter_data) # Should add many new stars (>1000) with new ids starting at 1000 assert len(out) > 1000 assert 1000 in out["id"] @@ -454,7 +430,41 @@ def test_add_jupiter_as_spoilers(): def test_add_jupiter_as_spoilers_no_jupiter(): # Should return stars unchanged if jupiter has zero length stars = Table({"row": [0], "col": [0], "mag": [8.0], "id": [1], "CLASS": [0]}) - out = jupiter.add_jupiter_as_lots_of_acq_spoilers( - "2025:220:12:00:00", stars, JupiterPositionTable.empty() + out = bright_object.add_bright_object_as_acq_spoilers( + "2025:220:12:00:00", stars, PlanetPositionTable.empty() ) assert out is stars + + +def test_check_for_close_planets_includes_non_jupiter(monkeypatch): + import chandra_aca.planets as planet_utils + + monkeypatch.setattr(bright_object.planets, "BRIGHT_PLANETS", ("venus", "mars")) + + def fake_get_planet_angular_sep(planet, ra, dec, time, observer_position): + return np.array([0.5]) + + def fake_get_planet_chandra_ccd_position( + planet, date, duration, att, ccd_pad, ephem_source + ): + assert ephem_source == "stk" + return PlanetPositionTable( + {"time": [CxoTime(date).secs], "row": [0.0], "col": [0.0]} + ) + + monkeypatch.setattr( + planet_utils, "get_planet_angular_sep", fake_get_planet_angular_sep + ) + monkeypatch.setattr( + planet_utils, + "get_planet_chandra_ccd_position", + fake_get_planet_chandra_ccd_position, + ) + + att = Quat(q=[-0.49963289, 0.25613709, -0.16664083, 0.81055018]) + out = bright_object.check_for_close_planets( + date="2025:220:12:00:00", duration=1000, att=att + ) + + assert set(out) == {"venus", "mars"} + assert all(isinstance(tbl, PlanetPositionTable) for tbl in out.values()) From 3afa4551ce5a71bab63f133862d5a95569915288 Mon Sep 17 00:00:00 2001 From: Jean Connelly Date: Tue, 26 May 2026 10:17:48 -0400 Subject: [PATCH 4/7] Simplify and move imports --- proseco/acq.py | 11 +++--- proseco/bright_object.py | 12 +++---- proseco/core.py | 74 +++++++++++++++++++--------------------- proseco/guide.py | 8 ++--- 4 files changed, 49 insertions(+), 56 deletions(-) diff --git a/proseco/acq.py b/proseco/acq.py index 7a3bb33e..b03994b6 100644 --- a/proseco/acq.py +++ b/proseco/acq.py @@ -10,17 +10,22 @@ from pathlib import Path from typing import TYPE_CHECKING +import astropy.units as u import numpy as np +from chandra_aca.planets import get_planet_mag_states from chandra_aca.star_probs import ( acq_success_prob, get_default_acq_prob_model_info, prob_n_acq, ) from chandra_aca.transform import mag_to_count_rate, pixels_to_yagzag, snr_mag_for_t_ccd +from cxotime import CxoTime from scipy import ndimage, stats from scipy.interpolate import interp1d from ska_helpers.utils import LazyDict +from proseco.bright_object import add_bright_object_as_acq_spoilers + from . import characteristics as ACA from . import characteristics_acq as ACQ from .core import ( @@ -257,12 +262,6 @@ def get_acq_catalog(obsid=0, **kwargs): # If bright planets are on-CCD, update stars with synthetic spoilers around # each bright object for acquisition selection. - import astropy.units as u - from chandra_aca.planets import get_planet_mag_states - from cxotime import CxoTime - - from proseco.bright_object import add_bright_object_as_acq_spoilers - for planet, planet_pos in acqs.planets.items(): if len(planet_pos) == 0: continue diff --git a/proseco/bright_object.py b/proseco/bright_object.py index 66559598..a31d0fca 100644 --- a/proseco/bright_object.py +++ b/proseco/bright_object.py @@ -4,6 +4,11 @@ import numpy as np from astropy.table import Table from chandra_aca import planets +from chandra_aca.planets import ( + convert_time_format_spk, + get_planet_angular_sep, + get_planet_chandra_ccd_position, +) from cxotime import CxoTime, CxoTimeLike from Quaternion import QuatLike @@ -19,13 +24,6 @@ def check_for_close_planets( att: QuatLike, tol=2.0, ) -> dict[str, "PlanetPositionTable"]: - from chandra_aca.planets import ( - convert_time_format_spk, - get_planet_angular_sep, - get_planet_chandra_ccd_position, - ) - from cxotime import CxoTime - date0 = CxoTime(date) if duration is None: time_secs = convert_time_format_spk(date0, "secs") diff --git a/proseco/core.py b/proseco/core.py index 02845c9a..3c6a68ef 100644 --- a/proseco/core.py +++ b/proseco/core.py @@ -9,12 +9,14 @@ import warnings from copy import copy from pathlib import Path -from typing import TYPE_CHECKING, Any, TypeAlias +from typing import Any, TypeAlias import agasc +import astropy.units as u import numpy as np from astropy.table import Column, Row, Table from chandra_aca.aca_image import AcaPsfLibrary +from chandra_aca.planets import PlanetPositionTable, get_planet_mag_states from chandra_aca.transform import ( count_rate_to_mag, mag_to_count_rate, @@ -23,10 +25,13 @@ yagzag_to_pixels, yagzag_to_radec, ) +from cxotime import CxoTime from mica.archive.aca_dark import get_dark_cal_id, get_dark_cal_image from Quaternion import Quat from scipy.interpolate import interp1d +from proseco.bright_object import check_for_close_planets + from . import characteristics as ACA # For testing this is used to cache fid tables for a detector @@ -37,9 +42,6 @@ get_dark_cal_image = functools.lru_cache(maxsize=6)(get_dark_cal_image) get_dark_cal_id = functools.lru_cache(maxsize=6)(get_dark_cal_id) -if TYPE_CHECKING: - from chandra_aca.planets import PlanetPositionTable - def to_python(val): try: @@ -631,53 +633,47 @@ def planets(self) -> dict[str, "PlanetPositionTable"]: if hasattr(self, "_planets"): return self._planets - if self.att is None: + if self.att is None or self.date is None: return {} - from proseco.bright_object import check_for_close_planets - self._planets = check_for_close_planets(self.date, self.duration, self.att) # Cache brightest state metadata on each planet table so downstream logic # can use a single source of truth without re-querying state files. - if self.date is not None: - import astropy.units as u - import numpy as np - from chandra_aca.planets import get_planet_mag_states - from cxotime import CxoTime - - duration = self.duration if self.duration is not None else 0.0 - for planet_name, planet_positions in self._planets.items(): - if len(planet_positions) == 0: - continue - - mag_states = get_planet_mag_states( - planet_name, - self.date, - CxoTime(self.date) + duration * u.s, - ) - if len(mag_states) == 0: - continue - - min_state_idx = np.argmin(mag_states["mag_start"]) - action_col = "label" if "label" in mag_states.colnames else "mag_action" - planet_positions.meta["brightest_mag_action"] = str( - mag_states[action_col][min_state_idx] - ) - planet_positions.meta["brightest_mag_start"] = float( - mag_states["mag_start"][min_state_idx] - ) - planet_positions.meta["brightest_mag_stop"] = float( - mag_states["mag_stop"][min_state_idx] - ) + duration = self.duration if self.duration is not None else 0.0 + for planet_name, planet_positions in self._planets.items(): + + # planet_positions may have zero length if the planet is within the + # check_for_close_planets tolerance but not actually on the CCD + if len(planet_positions) == 0: + continue + + mag_states = get_planet_mag_states( + planet_name, + self.date, + CxoTime(self.date) + duration * u.s, + ) + if len(mag_states) == 0: + continue + + # Min magnitude state is the brightest one + min_state_idx = np.argmin(mag_states["mag_start"]) + action_col = "label" if "label" in mag_states.colnames else "mag_action" + planet_positions.meta["brightest_mag_action"] = str( + mag_states[action_col][min_state_idx] + ) + planet_positions.meta["brightest_mag_start"] = float( + mag_states["mag_start"][min_state_idx] + ) + planet_positions.meta["brightest_mag_stop"] = float( + mag_states["mag_stop"][min_state_idx] + ) return self._planets @planets.setter def planets(self, value: dict[str, Any]) -> None: """Set planet positions dictionary.""" - from chandra_aca.planets import PlanetPositionTable - self._planets = {} if value is None: return diff --git a/proseco/guide.py b/proseco/guide.py index 48fb107b..f8195cd6 100644 --- a/proseco/guide.py +++ b/proseco/guide.py @@ -12,6 +12,10 @@ snr_mag_for_t_ccd, ) +from proseco.bright_object import ( + bright_object_distribution_check, + check_spoiled_by_bright_object, +) from proseco.characteristics import MonFunc if TYPE_CHECKING: @@ -648,8 +652,6 @@ def index_combinations(n, m): best_score = -1 best_cands = None - from proseco.bright_object import bright_object_distribution_check - # Weight the bright-object distribution check for planets on CCD that are in # partial or full mitigation states during the observation window. planets_for_distribution_check = [] @@ -1172,8 +1174,6 @@ def get_initial_guide_candidates(self): ) # Filter stars that are spoiled by any planet - from proseco.bright_object import check_spoiled_by_bright_object - for _, planet_positions in self.planets.items(): spoiled_by_planet, planet_rej = check_spoiled_by_bright_object( cand_guides, planet_positions From 86359bcfce310727de71e2977f04803f53563b4f Mon Sep 17 00:00:00 2001 From: Jean Connelly Date: Tue, 9 Jun 2026 13:18:54 -0400 Subject: [PATCH 5/7] Simplify --- proseco/acq.py | 3 --- proseco/core.py | 18 +++++++++--------- proseco/guide.py | 3 --- proseco/tests/test_jupiter.py | 7 +++---- 4 files changed, 12 insertions(+), 19 deletions(-) diff --git a/proseco/acq.py b/proseco/acq.py index b03994b6..5ede80e3 100644 --- a/proseco/acq.py +++ b/proseco/acq.py @@ -263,9 +263,6 @@ def get_acq_catalog(obsid=0, **kwargs): # If bright planets are on-CCD, update stars with synthetic spoilers around # each bright object for acquisition selection. for planet, planet_pos in acqs.planets.items(): - if len(planet_pos) == 0: - continue - duration = acqs.duration if acqs.duration is not None else 0 mag_states = get_planet_mag_states( planet, start=acqs.date, stop=CxoTime(acqs.date) + duration * u.s diff --git a/proseco/core.py b/proseco/core.py index 3c6a68ef..486ad398 100644 --- a/proseco/core.py +++ b/proseco/core.py @@ -629,25 +629,26 @@ class ACACatalogTable(BaseCatalogTable): @property def planets(self) -> dict[str, "PlanetPositionTable"]: - """Dictionary of planet positions keyed by planet name (lowercase).""" + """ + Dictionary of planet positions keyed by planet name (lowercase), + including only planets that are actually on the ACA CCD (len > 0). + """ if hasattr(self, "_planets"): return self._planets if self.att is None or self.date is None: return {} - self._planets = check_for_close_planets(self.date, self.duration, self.att) + # Only include planets that are actually on the CCD (len > 0) + self._planets = { + k: v for k, v in check_for_close_planets(self.date, self.duration, self.att).items() + if len(v) > 0 + } # Cache brightest state metadata on each planet table so downstream logic # can use a single source of truth without re-querying state files. duration = self.duration if self.duration is not None else 0.0 for planet_name, planet_positions in self._planets.items(): - - # planet_positions may have zero length if the planet is within the - # check_for_close_planets tolerance but not actually on the CCD - if len(planet_positions) == 0: - continue - mag_states = get_planet_mag_states( planet_name, self.date, @@ -655,7 +656,6 @@ def planets(self) -> dict[str, "PlanetPositionTable"]: ) if len(mag_states) == 0: continue - # Min magnitude state is the brightest one min_state_idx = np.argmin(mag_states["mag_start"]) action_col = "label" if "label" in mag_states.colnames else "mag_action" diff --git a/proseco/guide.py b/proseco/guide.py index f8195cd6..e3b13f68 100644 --- a/proseco/guide.py +++ b/proseco/guide.py @@ -656,9 +656,6 @@ def index_combinations(n, m): # partial or full mitigation states during the observation window. planets_for_distribution_check = [] for _, planet_positions in self.planets.items(): - if len(planet_positions) == 0: - continue - brightest_mag_action = planet_positions.meta.get("brightest_mag_action") # Allow tests that directly set _planets and omit metadata. diff --git a/proseco/tests/test_jupiter.py b/proseco/tests/test_jupiter.py index 2e421c9a..381639da 100644 --- a/proseco/tests/test_jupiter.py +++ b/proseco/tests/test_jupiter.py @@ -437,8 +437,6 @@ def test_add_jupiter_as_spoilers_no_jupiter(): def test_check_for_close_planets_includes_non_jupiter(monkeypatch): - import chandra_aca.planets as planet_utils - monkeypatch.setattr(bright_object.planets, "BRIGHT_PLANETS", ("venus", "mars")) def fake_get_planet_angular_sep(planet, ra, dec, time, observer_position): @@ -452,11 +450,12 @@ def fake_get_planet_chandra_ccd_position( {"time": [CxoTime(date).secs], "row": [0.0], "col": [0.0]} ) + # Patch on bright_object module since these are module-level bindings there monkeypatch.setattr( - planet_utils, "get_planet_angular_sep", fake_get_planet_angular_sep + bright_object, "get_planet_angular_sep", fake_get_planet_angular_sep ) monkeypatch.setattr( - planet_utils, + bright_object, "get_planet_chandra_ccd_position", fake_get_planet_chandra_ccd_position, ) From f123d0f0f9a2f6e1d77868c941a96d4a5b4fb54f Mon Sep 17 00:00:00 2001 From: Jean Connelly Date: Tue, 9 Jun 2026 14:14:44 -0400 Subject: [PATCH 6/7] Handle on CCD later in observation --- proseco/bright_object.py | 5 +++++ proseco/tests/test_jupiter.py | 10 ++++++++++ 2 files changed, 15 insertions(+) diff --git a/proseco/bright_object.py b/proseco/bright_object.py index a31d0fca..d930e974 100644 --- a/proseco/bright_object.py +++ b/proseco/bright_object.py @@ -283,6 +283,11 @@ def add_bright_object_as_acq_spoilers( # Bright object acq position acq_pos = get_bright_object_acq_pos(date, bright_object=bright_object) + # If the bright object is not on CCD close enough to acquisition start, + # skip adding synthetic spoilers for acquisition selection. + if acq_pos.row is None or acq_pos.col is None: + return stars + out = stars.copy() idincr = 0 for irow in np.arange(-505, 510, 5): diff --git a/proseco/tests/test_jupiter.py b/proseco/tests/test_jupiter.py index 381639da..f7d526ba 100644 --- a/proseco/tests/test_jupiter.py +++ b/proseco/tests/test_jupiter.py @@ -436,6 +436,16 @@ def test_add_jupiter_as_spoilers_no_jupiter(): assert out is stars +def test_add_jupiter_as_spoilers_delayed_entry(): + # If bright object first appears on CCD too long after acquisition start, + # no synthetic acquisition spoilers should be added. + date = "2025:220:12:00:00" + stars = StarsTable.empty() + delayed = Table({"time": [CxoTime(date).secs + 2500], "row": [100], "col": [100]}) + out = bright_object.add_bright_object_as_acq_spoilers(date, stars, delayed) + assert out is stars + + def test_check_for_close_planets_includes_non_jupiter(monkeypatch): monkeypatch.setattr(bright_object.planets, "BRIGHT_PLANETS", ("venus", "mars")) From 25f2c6f171471a1d73b78a77db01e6c2b8b08ff4 Mon Sep 17 00:00:00 2001 From: Jean Connelly Date: Tue, 9 Jun 2026 14:22:29 -0400 Subject: [PATCH 7/7] Fix formatting --- proseco/core.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/proseco/core.py b/proseco/core.py index 486ad398..ecfbbc39 100644 --- a/proseco/core.py +++ b/proseco/core.py @@ -641,7 +641,10 @@ def planets(self) -> dict[str, "PlanetPositionTable"]: # Only include planets that are actually on the CCD (len > 0) self._planets = { - k: v for k, v in check_for_close_planets(self.date, self.duration, self.att).items() + k: v + for k, v in check_for_close_planets( + self.date, self.duration, self.att + ).items() if len(v) > 0 }