diff --git a/RMS/Astrometry/AstrometryNet.py b/RMS/Astrometry/AstrometryNet.py index 921051303..c42e5b6c0 100644 --- a/RMS/Astrometry/AstrometryNet.py +++ b/RMS/Astrometry/AstrometryNet.py @@ -32,7 +32,7 @@ def astrometryNetSolveLocal(ff_file_path=None, img=None, mask=None, x_data=None, y_data=None, fov_w_range=None, fov_w_hint=None, max_stars=100, verbose=False, x_center=None, y_center=None, - lat=None, lon=None, jd=None, input_intensities=None): + lat=None, lon=None, jd=None, input_intensities=None, position_hint=None): """ Find an astrometric solution of X, Y image coordinates of stars detected on an image using the local installation of astrometry.net. @@ -52,6 +52,9 @@ def astrometryNetSolveLocal(ff_file_path=None, img=None, mask=None, x_data=None, lon: [float] Station longitude in degrees. Required for iterative matching. jd: [float] Julian date. Required for iterative matching. input_intensities: [ndarray] Star intensities for brightness-based matching. Optional. + position_hint: [tuple] Optional (ra_deg, dec_deg, radius_deg). When given, astrometry.net is + told roughly where the field is, collapsing the blind all-sky search to that patch -- + which lets narrow-FOV / faint cameras solve with far fewer stars. None = blind (default). Returns: [tuple] A tuple containing the following elements: @@ -207,6 +210,7 @@ def astrometryNetSolveLocal(ff_file_path=None, img=None, mask=None, x_data=None, scales = {14, 15, 16, 17, 18, 19} size_hint = None + avg_fov_w = fov_h = None # set below when a FOV estimate is available if fov_w_range is not None: @@ -324,6 +328,30 @@ def astrometryNetSolveLocal(ff_file_path=None, img=None, mask=None, x_data=None, ) ) + # Optional position hint: when the caller knows roughly where the camera points (e.g. a + # previous platepar, or a known station pointing), this collapses the blind all-sky search + # to a small patch, letting narrow-FOV / faint cameras solve with far fewer stars than a + # blind solve needs. None (the default) preserves the original blind behaviour. + position_hint_obj = None + if position_hint is not None: + ra_hint, dec_hint, radius_hint = position_hint + + # The local solver restricts the reference catalog to within radius_hint of the hint + # centre. If that radius is smaller than the image's own angular field, the outer stars + # get excluded and the quad match fails - a regression on wide-FOV cameras (e.g. a 15 deg + # hint on a ~64 deg fisheye drops everything past 15 deg from centre). Never let the hint + # be tighter than the field's circum-radius (half-diagonal, small margin). Only when a + # FOV estimate is available (it always is from autoFitPlatepar). + if avg_fov_w is not None and fov_h is not None: + field_radius = np.hypot(avg_fov_w, fov_h)/2.0*1.1 + radius_hint = max(float(radius_hint), field_radius) + + position_hint_obj = astrometry.PositionHint( + ra_deg=float(ra_hint), dec_deg=float(dec_hint), radius_deg=float(radius_hint)) + if verbose: + print("Using position hint: RA={:.2f} Dec={:.2f} radius={:.1f} deg (>= field radius)".format( + ra_hint, dec_hint, radius_hint)) + # If the solver.solve has the argument "stars", use a 2D array of stars instead of stars_xs and stars_ys solve_args = inspect.getfullargspec(solver.solve).args if "stars" in solve_args: @@ -332,7 +360,7 @@ def astrometryNetSolveLocal(ff_file_path=None, img=None, mask=None, x_data=None, solution = solver.solve( stars=star_data, size_hint=size_hint, - position_hint=None, + position_hint=position_hint_obj, solution_parameters=solution_parameters ) @@ -342,7 +370,7 @@ def astrometryNetSolveLocal(ff_file_path=None, img=None, mask=None, x_data=None, stars_xs=x_data, stars_ys=y_data, size_hint=size_hint, - position_hint=None, + position_hint=position_hint_obj, solution_parameters=solution_parameters ) @@ -451,9 +479,9 @@ def astrometryNetSolveLocal(ff_file_path=None, img=None, mask=None, x_data=None, def astrometryNetSolve(ff_file_path=None, img=None, mask=None, x_data=None, y_data=None, fov_w_range=None, fov_w_hint=None, max_stars=100, verbose=False, x_center=None, y_center=None, - lat=None, lon=None, jd=None, input_intensities=None): + lat=None, lon=None, jd=None, input_intensities=None, position_hint=None): """ Find an astrometric solution of X, Y image coordinates of stars detected on an image using the - local installation of astrometry.net. + local installation of astrometry.net or a compatible remote service. Keyword arguments: ff_file_path: [str] Path to the FF file to load. @@ -471,10 +499,14 @@ def astrometryNetSolve(ff_file_path=None, img=None, mask=None, x_data=None, y_da lon: [float] Station longitude in degrees. Required for iterative matching. jd: [float] Julian date. Required for iterative matching. input_intensities: [ndarray] Star intensities for brightness-based matching. Optional. + position_hint: [tuple] Optional (ra_deg, dec_deg, radius_deg). When given, astrometry.net is + told roughly where the field is, collapsing the blind all-sky search to that patch -- + which lets narrow-FOV / faint cameras solve with far fewer stars. None = blind (default). """ # Helper to try coordinate-only first, then fall back to image if available - def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_cen, y_cen): + def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_cen, y_cen, + position_hint): """Try coordinate-only solve first, fall back to image if that fails.""" # If we have coordinates, try coordinate-only first (faster, less bandwidth) @@ -484,7 +516,7 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce result = novaAstrometryNetSolve( ff_file_path=None, img=None, x_data=x_coords, y_data=y_coords, fov_w_range=fov_range, x_center=x_cen, y_center=y_cen, - api_url=api_url + api_url=api_url, position_hint=position_hint ) if result is not None: return result @@ -498,7 +530,7 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce return novaAstrometryNetSolve( ff_file_path=ff_path, img=image, x_data=None, y_data=None, fov_w_range=fov_range, x_center=x_cen, y_center=y_cen, - api_url=api_url + api_url=api_url, position_hint=position_hint ) return None @@ -506,7 +538,7 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce return novaAstrometryNetSolve( ff_file_path=ff_path, img=image, x_data=x_coords, y_data=y_coords, fov_w_range=fov_range, x_center=x_cen, y_center=y_cen, - api_url=api_url + api_url=api_url, position_hint=position_hint ) # If the local installation of astrometry.net is not available, use remote API @@ -518,7 +550,7 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce try: result = _tryRemoteSolve( PRIMARY_API_URL, ff_file_path, img, x_data, y_data, - fov_w_range, x_center, y_center + fov_w_range, x_center, y_center, position_hint ) if result is not None: return result @@ -529,7 +561,7 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce print(f"Trying fallback server: {FALLBACK_API_URL}") return _tryRemoteSolve( FALLBACK_API_URL, ff_file_path, img, x_data, y_data, - fov_w_range, x_center, y_center + fov_w_range, x_center, y_center, position_hint ) else: @@ -540,7 +572,7 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce ff_file_path=ff_file_path, img=img, mask=mask, x_data=x_data, y_data=y_data, fov_w_range=fov_w_range, fov_w_hint=fov_w_hint, max_stars=max_stars, verbose=verbose, x_center=x_center, y_center=y_center, - lat=lat, lon=lon, jd=jd, input_intensities=input_intensities + lat=lat, lon=lon, jd=jd, input_intensities=input_intensities, position_hint=position_hint ) # If local fails, try remote APIs @@ -554,7 +586,7 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce try: result = _tryRemoteSolve( PRIMARY_API_URL, ff_file_path, img, x_data, y_data, - fov_w_range, x_center, y_center + fov_w_range, x_center, y_center, position_hint ) if result is not None: return result @@ -565,7 +597,7 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce print(f"Trying fallback server: {FALLBACK_API_URL}") return _tryRemoteSolve( FALLBACK_API_URL, ff_file_path, img, x_data, y_data, - fov_w_range, x_center, y_center + fov_w_range, x_center, y_center, position_hint ) @@ -628,4 +660,4 @@ def _tryRemoteSolve(api_url, ff_path, image, x_coords, y_coords, fov_range, x_ce star['ra_deg'], star['dec_deg'], star['x_pix'], star['y_pix'])) else: - print("No solution found.") \ No newline at end of file + print("No solution found.") diff --git a/RMS/Astrometry/AstrometryNetNova.py b/RMS/Astrometry/AstrometryNetNova.py index 05582c15f..3108afbc4 100644 --- a/RMS/Astrometry/AstrometryNetNova.py +++ b/RMS/Astrometry/AstrometryNetNova.py @@ -282,7 +282,7 @@ def jobs_by_tag(self, tag, exact): def novaAstrometryNetSolve(ff_file_path=None, img=None, x_data=None, y_data=None, fov_w_range=None, - api_key=None, x_center=None, y_center=None, api_url=None): + api_key=None, x_center=None, y_center=None, api_url=None, position_hint=None): """ Find an astrometric solution of X, Y image coordinates of stars detected on an image using the nova.astrometry.net service or a compatible API. @@ -299,6 +299,7 @@ def novaAstrometryNetSolve(ff_file_path=None, img=None, x_data=None, y_data=None y_center: [float] Y coordinate of the image center. If not given, the image center will be used. api_url: [str] Custom API URL. None by default, in which case nova.astrometry.net will be used. Can be set to use alternative servers like 'https://astro.contrailcast.com/api/'. + position_hint: [tuple] Optional (ra_deg, dec_deg, radius_deg) search-position hint. Return: (ra, dec, orientation, scale, fov_w, fov_h, star_data): [tuple of floats] All in degrees, @@ -390,6 +391,13 @@ def _printWebLink(stat, first_status=None): kwargs['scale_upper'] = scale_upper kwargs['scale_units'] = 'degwidth' # FOV range is in degrees + # Restrict the remote all-sky search when an approximate field position is known + if position_hint is not None: + ra_hint, dec_hint, radius_hint = position_hint + kwargs['center_ra'] = float(ra_hint) + kwargs['center_dec'] = float(dec_hint) + kwargs['radius'] = float(radius_hint) + # Upload image or the list of stars if file_handle is not None: @@ -701,4 +709,4 @@ def _printWebLink(stat, first_status=None): # 'radius': 29.77070194214109, # 'dec': 75.03531583306331, # 'height_arcsec': 105022.27049595825, - # 'orientation': -75.0809053003417} \ No newline at end of file + # 'orientation': -75.0809053003417} diff --git a/RMS/Astrometry/AutoPlatepar.py b/RMS/Astrometry/AutoPlatepar.py index 4a5391a43..92999ebdc 100644 --- a/RMS/Astrometry/AutoPlatepar.py +++ b/RMS/Astrometry/AutoPlatepar.py @@ -330,7 +330,8 @@ def autoFitPlatepar(dir_path, config, catalog_stars, platepar_template=None, fwhm_mult=DEFAULT_BLEND_FWHM_MULT, wide_fov_search=False, final_catalog_stars=None, - verbose=True): + verbose=True, + position_hint=None): """ Automatically create a platepar from CALSTARS data in a directory. @@ -366,7 +367,11 @@ def autoFitPlatepar(dir_path, config, catalog_stars, platepar_template=None, wide_fov_search: [bool] If True, use a wide FOV search range (2° to 200°) instead of the config-based range. Used as fallback when the tight search fails. + final_catalog_stars: [ndarray] Optional catalog used for the final astrometric fit. verbose: [bool] Print progress information + position_hint: [tuple] Optional (ra_deg, dec_deg, radius_deg) forwarded to astrometry.net + as a search-position hint. Lets star-starved / narrow-FOV cameras solve + with fewer stars. None = blind all-sky search (default). Returns: platepar: [Platepar] Fitted platepar object, or None if fitting failed @@ -533,6 +538,7 @@ def autoFitPlatepar(dir_path, config, catalog_stars, platepar_template=None, lon=platepar.lon, jd=jd, input_intensities=input_intensities, + position_hint=position_hint, verbose=verbose ) @@ -550,6 +556,7 @@ def autoFitPlatepar(dir_path, config, catalog_stars, platepar_template=None, photometric_sigma=photometric_sigma, fwhm_mult=fwhm_mult, wide_fov_search=True, + position_hint=position_hint, verbose=verbose ) diff --git a/Tests/TestAstrometryNet.py b/Tests/TestAstrometryNet.py new file mode 100644 index 000000000..26f284f60 --- /dev/null +++ b/Tests/TestAstrometryNet.py @@ -0,0 +1,245 @@ +import datetime +import inspect +from types import SimpleNamespace + +import pytest + +np = pytest.importorskip("numpy") + +import RMS.Astrometry.AstrometryNet as AstrometryNet +import RMS.Astrometry.AstrometryNetNova as AstrometryNetNova +import RMS.Astrometry.AutoPlatepar as AutoPlatepar + + +class _NoMatchSolution(object): + def has_match(self): + return False + + +class _NewSolver(object): + call = None + + def __init__(self, index_files): + pass + + def solve(self, stars, size_hint, position_hint, solution_parameters): + _NewSolver.call = { + 'stars': stars, + 'position_hint': position_hint, + } + return _NoMatchSolution() + + +class _LegacySolver(object): + call = None + + def __init__(self, index_files): + pass + + def solve(self, stars_xs, stars_ys, size_hint, position_hint, solution_parameters): + _LegacySolver.call = { + 'stars_xs': stars_xs, + 'stars_ys': stars_ys, + 'position_hint': position_hint, + } + return _NoMatchSolution() + + +def _fakeAstrometry(solver_class): + return SimpleNamespace( + Solver=solver_class, + PositionHint=lambda **kwargs: SimpleNamespace(**kwargs), + SolutionParameters=lambda **kwargs: SimpleNamespace(**kwargs), + Action=SimpleNamespace(STOP='stop', CONTINUE='continue'), + series_4100=SimpleNamespace(index_files=lambda **kwargs: []), + ) + + +@pytest.mark.parametrize('solver_class', [_NewSolver, _LegacySolver]) +def test_local_solver_forwards_position_hint_for_supported_apis(monkeypatch, solver_class): + """ Position hints reach both the current and legacy astrometry solver APIs. """ + + solver_class.call = None + monkeypatch.setattr(AstrometryNet, 'astrometry', _fakeAstrometry(solver_class)) + + result = AstrometryNet.astrometryNetSolveLocal( + x_data=np.array([10.0, 20.0, 30.0]), + y_data=np.array([15.0, 25.0, 35.0]), + position_hint=(123, 45, 20), + ) + + assert result is None + assert solver_class.call is not None + hint = solver_class.call['position_hint'] + assert (hint.ra_deg, hint.dec_deg, hint.radius_deg) == (123.0, 45.0, 20.0) + + +def test_local_solver_preserves_blind_solve_default(monkeypatch): + """ Omitting a position hint keeps the original blind-solve behavior. """ + + _NewSolver.call = None + monkeypatch.setattr(AstrometryNet, 'astrometry', _fakeAstrometry(_NewSolver)) + + result = AstrometryNet.astrometryNetSolveLocal( + x_data=np.array([10.0, 20.0, 30.0]), + y_data=np.array([15.0, 25.0, 35.0]), + ) + + assert result is None + assert _NewSolver.call['position_hint'] is None + + +def test_remote_coordinate_and_image_fallbacks_preserve_position_hint(monkeypatch): + """ A failed coordinate solve keeps the hint when retrying with the image. """ + + calls = [] + + def fakeNovaSolve(**kwargs): + calls.append(kwargs) + return None if len(calls) == 1 else 'remote solution' + + monkeypatch.setattr(AstrometryNet, 'ASTROMETRY_NET_AVAILABLE', False) + monkeypatch.setattr(AstrometryNet, 'novaAstrometryNetSolve', fakeNovaSolve) + + position_hint = (123.0, 45.0, 20.0) + result = AstrometryNet.astrometryNetSolve( + img=np.zeros((10, 10), dtype=np.uint8), + x_data=np.array([1.0, 2.0]), + y_data=np.array([3.0, 4.0]), + position_hint=position_hint, + ) + + assert result == 'remote solution' + assert [call['position_hint'] for call in calls] == [position_hint, position_hint] + assert calls[0]['img'] is None + assert calls[1]['img'] is not None + + +def test_local_exception_remote_fallback_preserves_position_hint(monkeypatch): + """ A local solver error does not turn the remote retry into a blind solve. """ + + calls = [] + + def failLocalSolve(**kwargs): + raise RuntimeError('local failure') + + def fakeNovaSolve(**kwargs): + calls.append(kwargs) + return 'remote solution' + + monkeypatch.setattr(AstrometryNet, 'ASTROMETRY_NET_AVAILABLE', True) + monkeypatch.setattr(AstrometryNet, 'astrometryNetSolveLocal', failLocalSolve) + monkeypatch.setattr(AstrometryNet, 'novaAstrometryNetSolve', fakeNovaSolve) + + position_hint = (123.0, 45.0, 20.0) + result = AstrometryNet.astrometryNetSolve( + x_data=np.array([1.0, 2.0]), + y_data=np.array([3.0, 4.0]), + position_hint=position_hint, + ) + + assert result == 'remote solution' + assert calls[0]['position_hint'] == position_hint + + +def test_remote_solver_translates_position_hint_to_upload_fields(monkeypatch): + """ Remote submissions use the positional fields understood by Astrometry.net. """ + + class FakeClient(object): + upload_args = None + + def __init__(self, apiurl=None): + pass + + def login(self, api_key): + pass + + def upload(self, **kwargs): + FakeClient.upload_args = kwargs + return {'status': 'failure'} + + monkeypatch.setattr(AstrometryNetNova, 'Client', FakeClient) + + result = AstrometryNetNova.novaAstrometryNetSolve( + x_data=[1.0, 2.0], + y_data=[3.0, 4.0], + api_url='https://example.invalid/api/', + position_hint=(123, 45, 20), + ) + + assert result is False + assert FakeClient.upload_args['center_ra'] == 123.0 + assert FakeClient.upload_args['center_dec'] == 45.0 + assert FakeClient.upload_args['radius'] == 20.0 + + +def test_auto_fit_preserves_api_order_and_wide_retry_hint(monkeypatch): + """ Existing positional slots stay fixed and the wide retry keeps its position hint. """ + + parameters = list(inspect.signature(AutoPlatepar.autoFitPlatepar).parameters) + assert parameters[-4:] == [ + 'wide_fov_search', 'final_catalog_stars', 'verbose', 'position_hint'] + + ff_name = 'FF_test.bin' + star_data = np.array([ + [float(i), float(i + 1), 100.0, 1.0, 2.5, 0.0, 5.0, 0.0] + for i in range(12) + ]) + + monkeypatch.setattr(AutoPlatepar.os, 'listdir', lambda path: ['CALSTARS_test.txt']) + monkeypatch.setattr(AutoPlatepar.CALSTARS, 'readCALSTARS', + lambda path, name: ([(ff_name, star_data)], None)) + monkeypatch.setattr(AutoPlatepar, 'getMaskFile', lambda path, config: None) + monkeypatch.setattr(AutoPlatepar, 'filenameToDatetime', + lambda name: datetime.datetime(2026, 1, 1)) + monkeypatch.setattr(AutoPlatepar, 'date2JD', lambda *args: 2460000.0) + monkeypatch.setattr(AutoPlatepar, 'JD2HourAngle', lambda jd: 123.0) + + class FakePlatepar(object): + def __init__(self): + self.lat = 0.0 + self.lon = 0.0 + self.elev = 0.0 + self.X_res = 0 + self.Y_res = 0 + self.station_code = '' + + def addVignettingCoeff(self, use_flat=False): + pass + + monkeypatch.setattr(AutoPlatepar, 'Platepar', FakePlatepar) + + solver_hints = [] + monkeypatch.setattr(AutoPlatepar, 'astrometryNetSolve', + lambda **kwargs: solver_hints.append(kwargs['position_hint'])) + + recursive_calls = [] + retry_result = object() + + def fakeAutoFit(*args, **kwargs): + recursive_calls.append(kwargs) + return retry_result + + original_auto_fit = AutoPlatepar.autoFitPlatepar + monkeypatch.setattr(AutoPlatepar, 'autoFitPlatepar', fakeAutoFit) + + config = SimpleNamespace( + latitude=45.0, + longitude=16.0, + elevation=100.0, + width=1920, + height=1080, + stationID='TEST', + fov_w=75.0, + ) + position_hint = (123.0, 45.0, 20.0) + + result = original_auto_fit( + '/unused', config, np.empty((0, 3)), ff_name=ff_name, + verbose=False, position_hint=position_hint) + + assert result is retry_result + assert solver_hints == [position_hint] + assert len(recursive_calls) == 1 + assert recursive_calls[0]['wide_fov_search'] is True + assert recursive_calls[0]['position_hint'] == position_hint