From 452127227486b1cb2d87e945f9f3c582c7a29845 Mon Sep 17 00:00:00 2001 From: Anne Archibald Date: Sat, 10 Feb 2024 17:16:07 +0000 Subject: [PATCH 1/7] Use cached_property and types --- src/pint/observatory/__init__.py | 122 +++++++++++++++++++------------ src/pint/observatory/topo_obs.py | 78 ++++++++++---------- 2 files changed, 116 insertions(+), 84 deletions(-) diff --git a/src/pint/observatory/__init__.py b/src/pint/observatory/__init__.py index 895d46411..2935613fe 100644 --- a/src/pint/observatory/__init__.py +++ b/src/pint/observatory/__init__.py @@ -21,14 +21,17 @@ necessary. """ -from copy import deepcopy import os import textwrap from collections import defaultdict +from collections.abc import Callable +from copy import deepcopy from io import StringIO from pathlib import Path +from typing import Optional, Union import astropy.coordinates +import astropy.time import astropy.units as u import numpy as np from astropy.coordinates import EarthLocation @@ -97,7 +100,7 @@ def _load_gps_clock(): ) -def _load_bipm_clock(bipm_version): +def _load_bipm_clock(bipm_version: str): bipm_version = bipm_version.lower() if bipm_version not in _bipm_clock_versions: try: @@ -136,34 +139,40 @@ class Observatory: position. """ + fullname: str + aliases: list[str] + include_gps: bool + include_bipm: bool + bipm_version: str + # This is a dict containing all defined Observatory instances, # keyed on standard observatory name. - _registry = {} + _registry: dict[str, "Observatory"] = {} # This is a dict mapping any defined aliases to the corresponding # standard name. - _alias_map = {} + _alias_map: dict[str, str] = {} def __init__( self, - name, - fullname=None, - aliases=None, - include_gps=True, - include_bipm=True, - bipm_version=bipm_default, - overwrite=False, + name: str, + fullname: Optional[str] = None, + aliases: Optional[list[str]] = None, + include_gps: bool = True, + include_bipm: bool = True, + bipm_version: str = bipm_default, + overwrite: bool = False, ): - self._name = name.lower() - self._aliases = ( + self._name: str = name.lower() + self._aliases: list[str] = ( list(set(map(str.lower, aliases))) if aliases is not None else [] ) if aliases is not None: Observatory._add_aliases(self, aliases) - self.fullname = fullname if fullname is not None else name - self.include_gps = include_gps - self.include_bipm = include_bipm - self.bipm_version = bipm_version + self.fullname: str = fullname if fullname is not None else name + self.include_gps: bool = include_gps + self.include_bipm: bool = include_bipm + self.bipm_version: str = bipm_version if name.lower() in Observatory._registry: if not overwrite: @@ -175,16 +184,18 @@ def __init__( Observatory._register(self, name) @classmethod - def _register(cls, obs, name): - """Add an observatory to the registry using the specified name - (which will be converted to lower case). If an existing observatory + def _register(cls, obs: "Observatory", name: str): + """Add an observatory to the registry using the specified name (which will be converted to lower case). + + If an existing observatory of the same name exists, it will be replaced with the new one. The Observatory instance's name attribute will be updated for - consistency.""" + consistency. + """ cls._registry[name.lower()] = obs @classmethod - def _add_aliases(cls, obs, aliases): + def _add_aliases(cls, obs: "Observatory", aliases: list[str]): """Add aliases for the specified Observatory. Aliases should be given as a list. If any of the new aliases are already in use, they will be replaced. Aliases are not checked against the @@ -196,14 +207,17 @@ def _add_aliases(cls, obs, aliases): cls._alias_map[a.lower()] = obs.name @staticmethod - def gps_correction(t, limits="warn"): + def gps_correction(t: astropy.time.Time, limits: str = "warn"): """Compute the GPS clock corrections for times t.""" log.info("Applying GPS to UTC clock correction (~few nanoseconds)") _load_gps_clock() + assert _gps_clock is not None return _gps_clock.evaluate(t, limits=limits) @staticmethod - def bipm_correction(t, bipm_version=bipm_default, limits="warn"): + def bipm_correction( + t: astropy.time.Time, bipm_version: str = bipm_default, limits: str = "warn" + ): """Compute the GPS clock corrections for times t.""" log.info(f"Applying TT(TAI) to TT({bipm_version}) clock correction (~27 us)") tt2tai = 32.184 * 1e6 * u.us @@ -214,7 +228,7 @@ def bipm_correction(t, bipm_version=bipm_default, limits="warn"): ) @classmethod - def clear_registry(cls): + def clear_registry(cls) -> None: """Clear registry for ground-based observatories.""" cls._registry = {} cls._alias_map = {} @@ -229,7 +243,7 @@ def names(cls): return cls._registry.keys() @classmethod - def names_and_aliases(cls): + def names_and_aliases(cls) -> dict[str, list[str]]: """List all observatories and their aliases""" import pint.observatory.topo_obs # noqa import pint.observatory.special_locations # noqa @@ -241,15 +255,15 @@ def names_and_aliases(cls): # setter methods that update the registries appropriately. @property - def name(self): + def name(self) -> str: return self._name @property - def aliases(self): + def aliases(self) -> list[str]: return self._aliases @classmethod - def get(cls, name): + def get(cls, name: str) -> "Observatory": """Returns the Observatory instance for the specified name/alias. If the name has not been defined, an error will be raised. Aside @@ -303,9 +317,12 @@ def get(cls, name): # Any which raise NotImplementedError below must be implemented in # derived classes. - def earth_location_itrf(self, time=None): - """Returns observatory geocentric position as an astropy - EarthLocation object. For observatories where this is not + def earth_location_itrf( + self, time: Optional[astropy.time.Time] = None + ) -> Union[None, np.ndarray]: + """Returns observatory geocentric position as an astropy EarthLocation object. + + For observatories where this is not relevant, None can be returned. The location is in the International Terrestrial Reference Frame (ITRF). @@ -319,8 +336,9 @@ def earth_location_itrf(self, time=None): """ return None - def get_gcrs(self, t, ephem=None): - """Return position vector of observatory in GCRS + def get_gcrs(self, t: astropy.time.Time, ephem=None): + """Return position vector of observatory in GCRS. + t is an astropy.Time or array of astropy.Time objects ephem is a link to an ephemeris file. Needed for SSB observatory Returns a 3-vector of Quantities representing the position @@ -329,14 +347,17 @@ def get_gcrs(self, t, ephem=None): raise NotImplementedError @property - def timescale(self): - """Returns the timescale that TOAs from this observatory will be in, - once any clock corrections have been applied. This should be a + def timescale(self) -> str: + """Returns the timescale that TOAs from this observatory will be in, once any clock corrections have been applied. + + This should be a string suitable to be passed directly to the scale argument of astropy.time.Time().""" raise NotImplementedError - def clock_corrections(self, t, limits="warn"): + def clock_corrections( + self, t: astropy.time.Time, limits: str = "warn" + ) -> u.Quantity: """Compute clock corrections for a Time array. Given an array-valued Time, return the clock corrections @@ -356,7 +377,7 @@ def clock_corrections(self, t, limits="warn"): return corr - def last_clock_correction_mjd(self): + def last_clock_correction_mjd(self) -> float: """Return the MJD of the last available clock correction. Returns ``np.inf`` if no clock corrections are relevant. @@ -365,6 +386,7 @@ def last_clock_correction_mjd(self): if self.include_gps: _load_gps_clock() + assert _gps_clock is not None t = min(t, _gps_clock.last_correction_mjd()) if self.include_bipm: _load_bipm_clock(self.bipm_version) @@ -374,7 +396,13 @@ def last_clock_correction_mjd(self): ) return t - def get_TDBs(self, t, method="default", ephem=None, options=None): + def get_TDBs( + self, + t: astropy.time.Time, + method: Union[str, Callable] = "default", + ephem: Optional[str] = None, + options: Optional[dict] = None, + ): """This is a high level function for converting TOAs to TDB time scale. Different method can be applied to obtain the result. Current supported @@ -409,13 +437,13 @@ def get_TDBs(self, t, method="default", ephem=None, options=None): t = Time([t]) if t.scale == "tdb": return t - # Check the method. This pattern is from numpy minimize - meth = "_custom" if callable(method) else method.lower() if options is None: options = {} - if meth == "_custom": + if callable(method): options = dict(options) return method(t, **options) + else: + meth = method.lower() if meth == "default": return self._get_TDB_default(t, ephem) elif meth == "ephemeris": @@ -428,17 +456,17 @@ def get_TDBs(self, t, method="default", ephem=None, options=None): else: raise ValueError(f"Unknown method '{method}'.") - def _get_TDB_default(self, t, ephem): + def _get_TDB_default(self, t: astropy.time.Time, ephem): return t.tdb - def _get_TDB_ephem(self, t, ephem): + def _get_TDB_ephem(self, t: astropy.time.Time, ephem): """Read the ephem TDB-TT column. This column is provided by DE4XXt version of ephemeris. """ raise NotImplementedError - def posvel(self, t, ephem, group=None): + def posvel(self, t: astropy.time.Time, ephem, group=None): """Return observatory position and velocity for the given times. Position is relative to solar system barycenter; times are @@ -451,7 +479,7 @@ def posvel(self, t, ephem, group=None): def get_observatory( - name, include_gps=None, include_bipm=None, bipm_version=bipm_default + name: str, include_gps=None, include_bipm=None, bipm_version: str = bipm_default ): """Convenience function to get observatory object with options. diff --git a/src/pint/observatory/topo_obs.py b/src/pint/observatory/topo_obs.py index 1d7fe8868..33fc8b0e2 100644 --- a/src/pint/observatory/topo_obs.py +++ b/src/pint/observatory/topo_obs.py @@ -17,12 +17,15 @@ -------- :mod:`pint.observatory.special_locations` """ +import copy import json import os +from functools import cached_property from pathlib import Path -import copy +from typing import Optional import astropy.constants as c +import astropy.time import astropy.units as u import numpy as np from astropy.coordinates import EarthLocation @@ -36,9 +39,9 @@ NoClockCorrections, Observatory, bipm_default, + earth_location_distance, find_clock_file, get_observatory, - earth_location_distance, ) from pint.pulsar_mjd import Time from pint.solar_system_ephemerides import get_tdb_tt_ephem_geocenter, objPosVel_wrt_SSB @@ -149,36 +152,36 @@ class TopoObs(Observatory): def __init__( self, - name, + name: str, *, - fullname=None, - tempo_code=None, - itoa_code=None, - aliases=None, + fullname: Optional[str] = None, + tempo_code: Optional[str] = None, + itoa_code: Optional[str] = None, + aliases: Optional[list[str]] = None, location=None, itrf_xyz=None, - lat=None, - lon=None, + lat: Optional[float] = None, + lon: Optional[float] = None, height=None, - clock_file="", - clock_fmt="tempo", + clock_file: str = "", + clock_fmt: str = "tempo", clock_dir=None, - include_gps=True, - include_bipm=True, - bipm_version=bipm_default, + include_gps: bool = True, + include_bipm: bool = True, + bipm_version: str = bipm_default, origin=None, - overwrite=False, - bogus_last_correction=False, + overwrite: bool = False, + bogus_last_correction: bool = False, ): input_values = [lat is not None, lon is not None, height is not None] - if sum(input_values) > 0 and sum(input_values) < 3: + if any(input_values) and not all(input_values): raise ValueError("All of lat, lon, height are required for observatory") input_values = [ location is not None, itrf_xyz is not None, (lat is not None and lon is not None and height is not None), ] - if sum(input_values) == 0: + if not any(input_values): raise ValueError( f"EarthLocation, ITRF coordinates, or lat/lon/height are required for observatory '{name}'" ) @@ -209,11 +212,12 @@ def __init__( # Save clock file info, the data will be read only if clock # corrections for this site are requested. - self.clock_files = [clock_file] if isinstance(clock_file, str) else clock_file - self.clock_files = [c for c in self.clock_files if c != ""] - self.clock_fmt = clock_fmt + clock_files: list[str] = ( + [clock_file] if isinstance(clock_file, str) else clock_file + ) + self.clock_files: list[str] = [c for c in clock_files if c != ""] + self.clock_fmt: str = clock_fmt self.clock_dir = clock_dir - self._clock = None # The ClockFile objects, will be read on demand # If using TEMPO time.dat we need to know the 1-char tempo-style # observatory code. @@ -315,10 +319,9 @@ def separation(self, other, method="cartesian"): def earth_location_itrf(self, time=None): return self.location - def _load_clock_corrections(self): - if self._clock is not None: - return - self._clock = [] + @cached_property + def _clock(self) -> list: + clock = [] for cf in self.clock_files: if cf == "": continue @@ -326,7 +329,7 @@ def _load_clock_corrections(self): if isinstance(cf, dict): kwargs.update(cf) cf = kwargs.pop("name") - self._clock.append( + clock.append( find_clock_file( cf, format=self.clock_fmt, @@ -334,8 +337,11 @@ def _load_clock_corrections(self): **kwargs, ) ) + return clock - def clock_corrections(self, t, limits="warn"): + def clock_corrections( + self, t: astropy.time.Time, limits: str = "warn" + ) -> u.Quantity: """Compute the total clock corrections, Parameters @@ -344,17 +350,16 @@ def clock_corrections(self, t, limits="warn"): The time when the clock correcions are applied. """ - corr = super().clock_corrections(t, limits=limits) - # Read clock file if necessary - self._load_clock_corrections() + corr: u.Quantity = super().clock_corrections(t, limits=limits) if self._clock: log.info( f"Applying observatory clock corrections for observatory='{self.name}'." ) for clock in self._clock: corr += clock.evaluate(t, limits=limits) - elif self.clock_files: + # clock_files is not empty, but no clock corrections found + # FIXME: what if only some were found? msg = f"No clock corrections found for observatory {self.name} taken from file {self.clock_files}" if limits == "warn": log.warning(msg) @@ -365,19 +370,18 @@ def clock_corrections(self, t, limits="warn"): log.info(f"Observatory {self.name} requires no clock corrections.") return corr - def last_clock_correction_mjd(self): + def last_clock_correction_mjd(self) -> float: """Return the MJD of the last clock correction. Combines constraints based on Earth orientation parameters and on the available clock corrections specific to the telescope. """ t = super().last_clock_correction_mjd() - self._load_clock_corrections() for clock in self._clock: t = min(t, clock.last_correction_mjd()) return t - def _get_TDB_ephem(self, t, ephem): + def _get_TDB_ephem(self, t: astropy.time.Time, ephem): """Read the ephem TDB-TT column. This column is provided by DE4XXt version of ephemeris. This function is only @@ -406,7 +410,7 @@ def _get_TDB_ephem(self, t, ephem): location=self.earth_location_itrf(), ) - def get_gcrs(self, t, ephem=None): + def get_gcrs(self, t: astropy.time.Time, ephem=None): """Return position vector of TopoObs in GCRS Parameters @@ -423,7 +427,7 @@ def get_gcrs(self, t, ephem=None): ) return obs_geocenter_pv.pos - def posvel(self, t, ephem, group=None): + def posvel(self, t: astropy.time.Time, ephem, group=None): if t.isscalar: t = Time([t]) earth_pv = objPosVel_wrt_SSB("earth", t, ephem) From 32b28241ca048f9b3f53dd2707d439d064d78240 Mon Sep 17 00:00:00 2001 From: Anne Archibald Date: Sat, 10 Feb 2024 17:31:57 +0000 Subject: [PATCH 2/7] Fix python 3.8-incompatible syntax --- src/pint/observatory/__init__.py | 18 ++++++++-------- src/pint/observatory/topo_obs.py | 35 +++++++++++++++++++++++++++----- 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/src/pint/observatory/__init__.py b/src/pint/observatory/__init__.py index 2935613fe..65f4bcf5e 100644 --- a/src/pint/observatory/__init__.py +++ b/src/pint/observatory/__init__.py @@ -28,7 +28,7 @@ from copy import deepcopy from io import StringIO from pathlib import Path -from typing import Optional, Union +from typing import Optional, Union, List, Dict import astropy.coordinates import astropy.time @@ -140,31 +140,31 @@ class Observatory: """ fullname: str - aliases: list[str] + aliases: List[str] include_gps: bool include_bipm: bool bipm_version: str # This is a dict containing all defined Observatory instances, # keyed on standard observatory name. - _registry: dict[str, "Observatory"] = {} + _registry: Dict[str, "Observatory"] = {} # This is a dict mapping any defined aliases to the corresponding # standard name. - _alias_map: dict[str, str] = {} + _alias_map: Dict[str, str] = {} def __init__( self, name: str, fullname: Optional[str] = None, - aliases: Optional[list[str]] = None, + aliases: Optional[List[str]] = None, include_gps: bool = True, include_bipm: bool = True, bipm_version: str = bipm_default, overwrite: bool = False, ): self._name: str = name.lower() - self._aliases: list[str] = ( + self._aliases: List[str] = ( list(set(map(str.lower, aliases))) if aliases is not None else [] ) if aliases is not None: @@ -195,7 +195,7 @@ def _register(cls, obs: "Observatory", name: str): cls._registry[name.lower()] = obs @classmethod - def _add_aliases(cls, obs: "Observatory", aliases: list[str]): + def _add_aliases(cls, obs: "Observatory", aliases: List[str]): """Add aliases for the specified Observatory. Aliases should be given as a list. If any of the new aliases are already in use, they will be replaced. Aliases are not checked against the @@ -243,7 +243,7 @@ def names(cls): return cls._registry.keys() @classmethod - def names_and_aliases(cls) -> dict[str, list[str]]: + def names_and_aliases(cls) -> Dict[str, List[str]]: """List all observatories and their aliases""" import pint.observatory.topo_obs # noqa import pint.observatory.special_locations # noqa @@ -259,7 +259,7 @@ def name(self) -> str: return self._name @property - def aliases(self) -> list[str]: + def aliases(self) -> List[str]: return self._aliases @classmethod diff --git a/src/pint/observatory/topo_obs.py b/src/pint/observatory/topo_obs.py index 33fc8b0e2..4afc341e9 100644 --- a/src/pint/observatory/topo_obs.py +++ b/src/pint/observatory/topo_obs.py @@ -22,7 +22,7 @@ import os from functools import cached_property from pathlib import Path -from typing import Optional +from typing import Optional, Union, List import astropy.constants as c import astropy.time @@ -150,6 +150,31 @@ class TopoObs(Observatory): """ + tempo_code: Optional[str] + """One-character TEMPO code.""" + itoa_code: Optional[str] + """Two-character ITOA code.""" + location: EarthLocation + """Location of the observatory.""" + clock_files: List[str] + """List of files to read for clock corrections. If empty, no clock corrections are applied.""" + clock_fmt: str + """Format of the clock files. + + See :class:`pint.observatory.clock_file.ClockFile` for allowed values. + """ + bogus_last_correction: bool + """Clock correction files include a bogus last correction. + + This is common with TEMPO/TEMPO2 clock files since neither program does + a good job with times past the end ot the table. It makes detecting values + past the end of real calibration difficult if it's not marked as bogus. + """ + clock_dir: Optional[Union[str, Path]] + """Where to look for the clock files.""" + origin: Optional[str] + """Documentation of the origin/author/date for the information.""" + def __init__( self, name: str, @@ -157,7 +182,7 @@ def __init__( fullname: Optional[str] = None, tempo_code: Optional[str] = None, itoa_code: Optional[str] = None, - aliases: Optional[list[str]] = None, + aliases: Optional[List[str]] = None, location=None, itrf_xyz=None, lat: Optional[float] = None, @@ -169,7 +194,7 @@ def __init__( include_gps: bool = True, include_bipm: bool = True, bipm_version: str = bipm_default, - origin=None, + origin: Optional[str] = None, overwrite: bool = False, bogus_last_correction: bool = False, ): @@ -212,10 +237,10 @@ def __init__( # Save clock file info, the data will be read only if clock # corrections for this site are requested. - clock_files: list[str] = ( + clock_files: List[str] = ( [clock_file] if isinstance(clock_file, str) else clock_file ) - self.clock_files: list[str] = [c for c in clock_files if c != ""] + self.clock_files: List[str] = [c for c in clock_files if c != ""] self.clock_fmt: str = clock_fmt self.clock_dir = clock_dir From 10a7248aa4dd78bc4c4a7c9dc4bac6ca9847ee8f Mon Sep 17 00:00:00 2001 From: Anne Archibald Date: Sat, 10 Feb 2024 17:46:17 +0000 Subject: [PATCH 3/7] Fill out typing more --- src/pint/observatory/__init__.py | 27 +++++++++++++-------------- src/pint/observatory/topo_obs.py | 32 +++++++++++++++----------------- 2 files changed, 28 insertions(+), 31 deletions(-) diff --git a/src/pint/observatory/__init__.py b/src/pint/observatory/__init__.py index 65f4bcf5e..8803b655b 100644 --- a/src/pint/observatory/__init__.py +++ b/src/pint/observatory/__init__.py @@ -39,7 +39,7 @@ from pint.config import runtimefile from pint.pulsar_mjd import Time -from pint.utils import interesting_lines +from pint.utils import interesting_lines, PosVel # Include any files that define observatories here. This will start # with the standard distribution files, then will read any system- or @@ -90,7 +90,7 @@ class ClockCorrectionOutOfRange(ClockCorrectionError): _bipm_clock_versions = {} -def _load_gps_clock(): +def _load_gps_clock() -> None: global _gps_clock if _gps_clock is None: log.info("Loading global GPS clock file") @@ -100,7 +100,7 @@ def _load_gps_clock(): ) -def _load_bipm_clock(bipm_version: str): +def _load_bipm_clock(bipm_version: str) -> None: bipm_version = bipm_version.lower() if bipm_version not in _bipm_clock_versions: try: @@ -140,7 +140,6 @@ class Observatory: """ fullname: str - aliases: List[str] include_gps: bool include_bipm: bool bipm_version: str @@ -184,7 +183,7 @@ def __init__( Observatory._register(self, name) @classmethod - def _register(cls, obs: "Observatory", name: str): + def _register(cls, obs: "Observatory", name: str) -> None: """Add an observatory to the registry using the specified name (which will be converted to lower case). If an existing observatory @@ -195,7 +194,7 @@ def _register(cls, obs: "Observatory", name: str): cls._registry[name.lower()] = obs @classmethod - def _add_aliases(cls, obs: "Observatory", aliases: List[str]): + def _add_aliases(cls, obs: "Observatory", aliases: List[str]) -> None: """Add aliases for the specified Observatory. Aliases should be given as a list. If any of the new aliases are already in use, they will be replaced. Aliases are not checked against the @@ -207,7 +206,7 @@ def _add_aliases(cls, obs: "Observatory", aliases: List[str]): cls._alias_map[a.lower()] = obs.name @staticmethod - def gps_correction(t: astropy.time.Time, limits: str = "warn"): + def gps_correction(t: astropy.time.Time, limits: str = "warn") -> u.Quantity: """Compute the GPS clock corrections for times t.""" log.info("Applying GPS to UTC clock correction (~few nanoseconds)") _load_gps_clock() @@ -217,7 +216,7 @@ def gps_correction(t: astropy.time.Time, limits: str = "warn"): @staticmethod def bipm_correction( t: astropy.time.Time, bipm_version: str = bipm_default, limits: str = "warn" - ): + ) -> u.Quantity: """Compute the GPS clock corrections for times t.""" log.info(f"Applying TT(TAI) to TT({bipm_version}) clock correction (~27 us)") tt2tai = 32.184 * 1e6 * u.us @@ -466,7 +465,7 @@ def _get_TDB_ephem(self, t: astropy.time.Time, ephem): """ raise NotImplementedError - def posvel(self, t: astropy.time.Time, ephem, group=None): + def posvel(self, t: astropy.time.Time, ephem, group=None) -> PosVel: """Return observatory position and velocity for the given times. Position is relative to solar system barycenter; times are @@ -519,14 +518,14 @@ def get_observatory( return Observatory.get(name) -def earth_location_distance(loc1, loc2): +def earth_location_distance(loc1: EarthLocation, loc2: EarthLocation) -> u.Quantity: """Compute the distance between two EarthLocations.""" return ( sum((u.Quantity(loc1.to_geocentric()) - u.Quantity(loc2.to_geocentric())) ** 2) ) ** 0.5 -def compare_t2_observatories_dat(t2dir=None): +def compare_t2_observatories_dat(t2dir: Optional[str] = None) -> Dict[str, List[Dict]]: """Read a tempo2 observatories.dat file and compare with PINT Produces a report including lines that can be added to PINT's @@ -617,7 +616,7 @@ def compare_t2_observatories_dat(t2dir=None): return report -def compare_tempo_obsys_dat(tempodir=None): +def compare_tempo_obsys_dat(tempodir: Optional[str] = None) -> Dict[str, List[Dict]]: """Read a tempo obsys.dat file and compare with PINT. Produces a report including lines that can be added to PINT's @@ -657,8 +656,8 @@ def compare_tempo_obsys_dat(tempodir=None): y = float(line_io.read(15)) z = float(line_io.read(15)) line_io.read(2) - icoord = line_io.read(1).strip() - icoord = int(icoord) if icoord else 0 + icoord_str = line_io.read(1).strip() + icoord = int(icoord_str) if icoord_str else 0 line_io.read(2) obsnam = line_io.read(20).strip().lower() tempo_code = line_io.read(1) diff --git a/src/pint/observatory/topo_obs.py b/src/pint/observatory/topo_obs.py index 4afc341e9..0b810aa12 100644 --- a/src/pint/observatory/topo_obs.py +++ b/src/pint/observatory/topo_obs.py @@ -45,7 +45,7 @@ ) from pint.pulsar_mjd import Time from pint.solar_system_ephemerides import get_tdb_tt_ephem_geocenter, objPosVel_wrt_SSB -from pint.utils import has_astropy_unit, open_or_use +from pint.utils import has_astropy_unit, open_or_use, PosVel # environment variables that can override clock location and observatory location pint_obs_env_var = "PINT_OBS_OVERRIDE" @@ -277,7 +277,7 @@ def __init__( overwrite=overwrite, ) - def __repr__(self): + def __repr__(self) -> str: aliases = [f"'{x}'" for x in self.aliases] origin = ( f"{self.fullname}\n{self.origin}" @@ -309,7 +309,7 @@ def get_json(self): """Return as a JSON string""" return json.dumps(self.get_dict()) - def separation(self, other, method="cartesian"): + def separation(self, other: "TopoObs", method: str = "cartesian"): """Return separation between two TopoObs objects Parameters @@ -341,7 +341,7 @@ def separation(self, other, method="cartesian"): ) return (c.R_earth * dsigma).to(u.m, equivalencies=u.dimensionless_angles()) - def earth_location_itrf(self, time=None): + def earth_location_itrf(self, time=None) -> EarthLocation: return self.location @cached_property @@ -364,9 +364,7 @@ def _clock(self) -> list: ) return clock - def clock_corrections( - self, t: astropy.time.Time, limits: str = "warn" - ) -> u.Quantity: + def clock_corrections(self, t: Time, limits: str = "warn") -> u.Quantity: """Compute the total clock corrections, Parameters @@ -406,7 +404,7 @@ def last_clock_correction_mjd(self) -> float: t = min(t, clock.last_correction_mjd()) return t - def _get_TDB_ephem(self, t: astropy.time.Time, ephem): + def _get_TDB_ephem(self, t: Time, ephem) -> Time: """Read the ephem TDB-TT column. This column is provided by DE4XXt version of ephemeris. This function is only @@ -418,8 +416,8 @@ def _get_TDB_ephem(self, t: astropy.time.Time, ephem): # Topocenter to Geocenter # Since earth velocity is not going to change a lot in 3ms. The # differences between TT and TDB can be ignored. - earth_pv = objPosVel_wrt_SSB("earth", t.tdb, ephem) - obs_geocenter_pv = gcrs_posvel_from_itrf( + earth_pv: PosVel = objPosVel_wrt_SSB("earth", t.tdb, ephem) + obs_geocenter_pv: PosVel = gcrs_posvel_from_itrf( self.earth_location_itrf(), t, obsname=self.name ) # NOTE @@ -447,22 +445,22 @@ def get_gcrs(self, t: astropy.time.Time, ephem=None): np.array a 3-vector of Quantities representing the position in GCRS coordinates. """ - obs_geocenter_pv = gcrs_posvel_from_itrf( + obs_geocenter_pv: PosVel = gcrs_posvel_from_itrf( self.earth_location_itrf(), t, obsname=self.name ) return obs_geocenter_pv.pos - def posvel(self, t: astropy.time.Time, ephem, group=None): + def posvel(self, t: astropy.time.Time, ephem, group=None) -> PosVel: if t.isscalar: t = Time([t]) - earth_pv = objPosVel_wrt_SSB("earth", t, ephem) - obs_geocenter_pv = gcrs_posvel_from_itrf( + earth_pv: PosVel = objPosVel_wrt_SSB("earth", t, ephem) + obs_geocenter_pv: PosVel = gcrs_posvel_from_itrf( self.earth_location_itrf(), t, obsname=self.name ) return obs_geocenter_pv + earth_pv -def export_all_clock_files(directory): +def export_all_clock_files(directory: Union[str, Path]) -> None: """Export all clock files PINT is using. This will export all the clock files PINT is using - every clock file used @@ -494,7 +492,7 @@ def export_all_clock_files(directory): clock.export(directory / Path(clock.filename).name) -def load_observatories(filename=observatories_json, overwrite=False): +def load_observatories(filename=observatories_json, overwrite: bool = False) -> None: """Load observatory definitions from JSON and create :class:`pint.observatory.topo_obs.TopoObs` objects, registering them Set `overwrite` to ``True`` if you want to re-read a file with updated definitions. @@ -528,7 +526,7 @@ def load_observatories(filename=observatories_json, overwrite=False): TopoObs(name=obsname, **obsdict) -def load_observatories_from_usual_locations(clear=False): +def load_observatories_from_usual_locations(clear: bool = False) -> None: """Load observatories from the default JSON file as well as ``$PINT_OBS_OVERRIDE``, optionally clearing the registry Running with ``clear=True`` will return PINT to the state it is on import. From f39772687f24604cfdee26dd1eb95b28f8502586 Mon Sep 17 00:00:00 2001 From: Anne Archibald Date: Sat, 10 Feb 2024 18:16:46 +0000 Subject: [PATCH 4/7] Improve attribute documentation --- src/pint/observatory/__init__.py | 13 +++++++++++++ src/pint/observatory/topo_obs.py | 4 ++-- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/src/pint/observatory/__init__.py b/src/pint/observatory/__init__.py index 8803b655b..8a5663b7d 100644 --- a/src/pint/observatory/__init__.py +++ b/src/pint/observatory/__init__.py @@ -140,9 +140,13 @@ class Observatory: """ fullname: str + """Full human-readable name of the observatory.""" include_gps: bool + """Whether to include GPS clock corrections.""" include_bipm: bool + """Whether to include BIPM clock corrections.""" bipm_version: str + """Version of the BIPM clock file to use.""" # This is a dict containing all defined Observatory instances, # keyed on standard observatory name. @@ -255,10 +259,19 @@ def names_and_aliases(cls) -> Dict[str, List[str]]: @property def name(self) -> str: + """Short name of the observatory. + + This is the name used in TOA files and in the observatory registry. + """ return self._name @property def aliases(self) -> List[str]: + """List of aliases for the observatory. + + These are short names also used to specify this observatory. + Includes ITOA and TEMPO codes, and any other common names. + """ return self._aliases @classmethod diff --git a/src/pint/observatory/topo_obs.py b/src/pint/observatory/topo_obs.py index 0b810aa12..4a247e701 100644 --- a/src/pint/observatory/topo_obs.py +++ b/src/pint/observatory/topo_obs.py @@ -306,11 +306,11 @@ def get_dict(self): return {self.name: output} def get_json(self): - """Return as a JSON string""" + """Return as a JSON string.""" return json.dumps(self.get_dict()) def separation(self, other: "TopoObs", method: str = "cartesian"): - """Return separation between two TopoObs objects + """Return separation between two TopoObs objects. Parameters ---------- From 6da627a1ecae2e2a82410da87dd647a66b91e34e Mon Sep 17 00:00:00 2001 From: Anne Archibald Date: Wed, 21 Feb 2024 18:50:07 +0000 Subject: [PATCH 5/7] Additional type annotations --- src/pint/observatory/__init__.py | 33 +++++++++++++++++--------------- src/pint/observatory/topo_obs.py | 18 ++++++++--------- 2 files changed, 27 insertions(+), 24 deletions(-) diff --git a/src/pint/observatory/__init__.py b/src/pint/observatory/__init__.py index 8a5663b7d..ad889ded1 100644 --- a/src/pint/observatory/__init__.py +++ b/src/pint/observatory/__init__.py @@ -28,7 +28,7 @@ from copy import deepcopy from io import StringIO from pathlib import Path -from typing import Optional, Union, List, Dict +from typing import Optional, Union, List, Dict, Literal import astropy.coordinates import astropy.time @@ -348,7 +348,7 @@ def earth_location_itrf( """ return None - def get_gcrs(self, t: astropy.time.Time, ephem=None): + def get_gcrs(self, t: astropy.time.Time, ephem: Optional[str] = None): """Return position vector of observatory in GCRS. t is an astropy.Time or array of astropy.Time objects @@ -468,17 +468,17 @@ def get_TDBs( else: raise ValueError(f"Unknown method '{method}'.") - def _get_TDB_default(self, t: astropy.time.Time, ephem): + def _get_TDB_default(self, t: astropy.time.Time, ephem: Optional[str]): return t.tdb - def _get_TDB_ephem(self, t: astropy.time.Time, ephem): + def _get_TDB_ephem(self, t: astropy.time.Time, ephem: Optional[str]): """Read the ephem TDB-TT column. This column is provided by DE4XXt version of ephemeris. """ raise NotImplementedError - def posvel(self, t: astropy.time.Time, ephem, group=None) -> PosVel: + def posvel(self, t: astropy.time.Time, ephem: Optional[str], group=None) -> PosVel: """Return observatory position and velocity for the given times. Position is relative to solar system barycenter; times are @@ -491,7 +491,10 @@ def posvel(self, t: astropy.time.Time, ephem, group=None) -> PosVel: def get_observatory( - name: str, include_gps=None, include_bipm=None, bipm_version: str = bipm_default + name: str, + include_gps: Optional[bool] = None, + include_bipm: Optional[bool] = None, + bipm_version: str = bipm_default, ): """Convenience function to get observatory object with options. @@ -753,7 +756,7 @@ def convert_angle(x): return report -def list_last_correction_mjds(): +def list_last_correction_mjds() -> None: """Print out a list of the last MJD each clock correction is good for. Each observatory lists the clock files it uses and their last dates, @@ -784,7 +787,7 @@ def list_last_correction_mjds(): print(f" {c.friendly_name:<20} MISSING") -def update_clock_files(bipm_versions=None): +def update_clock_files(bipm_versions: Optional[list[str]] = None) -> None: """Obtain an up-to-date version of all clock files. This up-to-date version will be stored in the Astropy cache; @@ -826,13 +829,13 @@ def update_clock_files(bipm_versions=None): # Both topo_obs and special_locations need this def find_clock_file( - name, - format, - bogus_last_correction=False, - url_base=None, - clock_dir=None, - valid_beyond_ends=False, -): + name: str, + format: Literal["tempo", "tempo2"], + bogus_last_correction: bool = False, + url_base: Optional[str] = None, + clock_dir: Union[str, Path, None] = None, + valid_beyond_ends: bool = False, +) -> "ClockFile": """Locate and return a ClockFile in one of several places. PINT looks for clock files in three places, in order: diff --git a/src/pint/observatory/topo_obs.py b/src/pint/observatory/topo_obs.py index 4a247e701..53ad6362e 100644 --- a/src/pint/observatory/topo_obs.py +++ b/src/pint/observatory/topo_obs.py @@ -22,7 +22,7 @@ import os from functools import cached_property from pathlib import Path -from typing import Optional, Union, List +from typing import Optional, Union, List, Any import astropy.constants as c import astropy.time @@ -183,14 +183,14 @@ def __init__( tempo_code: Optional[str] = None, itoa_code: Optional[str] = None, aliases: Optional[List[str]] = None, - location=None, + location: Optional[EarthLocation] = None, itrf_xyz=None, lat: Optional[float] = None, lon: Optional[float] = None, height=None, clock_file: str = "", clock_fmt: str = "tempo", - clock_dir=None, + clock_dir: Union[str, Path, None] = None, include_gps: bool = True, include_bipm: bool = True, bipm_version: str = bipm_default, @@ -287,10 +287,10 @@ def __repr__(self) -> str: return f"TopoObs('{self.name}' ({','.join(aliases)}) at [{self.location.x}, {self.location.y} {self.location.z}]:\n{origin})" @property - def timescale(self): + def timescale(self) -> str: return "utc" - def get_dict(self): + def get_dict(self) -> dict[str, dict[str, Any]]: """Return as a dict with limited/changed info""" # start with the default __dict__ # copy some attributes to rename them and remove those that aren't needed for initialization @@ -305,11 +305,11 @@ def get_dict(self): output["itrf_xyz"] = [x.to_value(u.m) for x in self.location.geocentric] return {self.name: output} - def get_json(self): + def get_json(self) -> str: """Return as a JSON string.""" return json.dumps(self.get_dict()) - def separation(self, other: "TopoObs", method: str = "cartesian"): + def separation(self, other: "TopoObs", method: str = "cartesian") -> u.Quantity: """Return separation between two TopoObs objects. Parameters @@ -433,7 +433,7 @@ def _get_TDB_ephem(self, t: Time, ephem) -> Time: location=self.earth_location_itrf(), ) - def get_gcrs(self, t: astropy.time.Time, ephem=None): + def get_gcrs(self, t: astropy.time.Time, ephem: Optional[str] = None): """Return position vector of TopoObs in GCRS Parameters @@ -450,7 +450,7 @@ def get_gcrs(self, t: astropy.time.Time, ephem=None): ) return obs_geocenter_pv.pos - def posvel(self, t: astropy.time.Time, ephem, group=None) -> PosVel: + def posvel(self, t: astropy.time.Time, ephem: Optional[str], group=None) -> PosVel: if t.isscalar: t = Time([t]) earth_pv: PosVel = objPosVel_wrt_SSB("earth", t, ephem) From 23f5d6d6025d6af721328d7e751474f5f745a839 Mon Sep 17 00:00:00 2001 From: Anne Archibald Date: Wed, 21 Feb 2024 18:54:43 +0000 Subject: [PATCH 6/7] Fix list->List --- src/pint/observatory/__init__.py | 2 +- src/pint/observatory/topo_obs.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pint/observatory/__init__.py b/src/pint/observatory/__init__.py index ad889ded1..bccb40d53 100644 --- a/src/pint/observatory/__init__.py +++ b/src/pint/observatory/__init__.py @@ -787,7 +787,7 @@ def list_last_correction_mjds() -> None: print(f" {c.friendly_name:<20} MISSING") -def update_clock_files(bipm_versions: Optional[list[str]] = None) -> None: +def update_clock_files(bipm_versions: Optional[List[str]] = None) -> None: """Obtain an up-to-date version of all clock files. This up-to-date version will be stored in the Astropy cache; diff --git a/src/pint/observatory/topo_obs.py b/src/pint/observatory/topo_obs.py index 53ad6362e..8cbe1ddc0 100644 --- a/src/pint/observatory/topo_obs.py +++ b/src/pint/observatory/topo_obs.py @@ -404,7 +404,7 @@ def last_clock_correction_mjd(self) -> float: t = min(t, clock.last_correction_mjd()) return t - def _get_TDB_ephem(self, t: Time, ephem) -> Time: + def _get_TDB_ephem(self, t: Time, ephem: Optional[str]) -> Time: """Read the ephem TDB-TT column. This column is provided by DE4XXt version of ephemeris. This function is only From 6fda954b49bcf79c9d25c3ffa59ea9dc2b6baf37 Mon Sep 17 00:00:00 2001 From: Anne Archibald Date: Wed, 21 Feb 2024 18:58:17 +0000 Subject: [PATCH 7/7] dict -> Dict --- src/pint/observatory/topo_obs.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pint/observatory/topo_obs.py b/src/pint/observatory/topo_obs.py index 8cbe1ddc0..3e47d9b5b 100644 --- a/src/pint/observatory/topo_obs.py +++ b/src/pint/observatory/topo_obs.py @@ -22,7 +22,7 @@ import os from functools import cached_property from pathlib import Path -from typing import Optional, Union, List, Any +from typing import Optional, Union, List, Any, Dict import astropy.constants as c import astropy.time @@ -290,7 +290,7 @@ def __repr__(self) -> str: def timescale(self) -> str: return "utc" - def get_dict(self) -> dict[str, dict[str, Any]]: + def get_dict(self) -> Dict[str, Dict[str, Any]]: """Return as a dict with limited/changed info""" # start with the default __dict__ # copy some attributes to rename them and remove those that aren't needed for initialization