Python >= 3.7 type annotation

Author: scivision

setup.cfg

@@ -23,7 +23,7 @@ long_description = file: README.md
 long_description_content_type = text/markdown
 
 [options]
-python_requires = >= 3.6.2
+python_requires = >= 3.7
 include_package_data = True
 zip_safe = False
 packages = find:

src/georinex/base.py

@@ -1,7 +1,7 @@
+from __future__ import annotations
+import typing as T
 from pathlib import Path
 import xarray
-from typing import Union, Tuple, Dict, Sequence
-from typing.io import TextIO
 from datetime import datetime, timedelta
 import logging
 
@@ -18,18 +18,18 @@
 
 
 def load(
-    rinexfn: Union[TextIO, str, Path],
+    rinexfn: T.TextIO | str | Path,
     out: Path = None,
-    use: Sequence[str] = None,
-    tlim: Tuple[datetime, datetime] = None,
+    use: list[str] = None,
+    tlim: tuple[datetime, datetime] = None,
     useindicators: bool = False,
-    meas: Sequence[str] = None,
+    meas: list[str] = None,
     verbose: bool = False,
     *,
     overwrite: bool = False,
     fast: bool = True,
-    interval: Union[float, int, timedelta] = None,
-) -> Union[xarray.Dataset, Dict[str, xarray.Dataset]]:
+    interval: float | int | timedelta = None,
+) -> xarray.Dataset | dict[str, xarray.Dataset]:
     """
     Reads OBS, NAV in RINEX 2.x and 3.x
 
@@ -67,9 +67,7 @@ def load(
         )
         return None
 
-    if info["rinextype"] == "sp3":
-        return load_sp3(rinexfn, outfn)
-    elif info["rinextype"] == "nav":
+    if info["rinextype"] == "nav":
         return rinexnav(rinexfn, outfn, use=use, tlim=tlim, overwrite=overwrite)
     elif info["rinextype"] == "obs":
         return rinexobs(
@@ -84,6 +82,11 @@ def load(
             fast=fast,
             interval=interval,
         )
+
+    assert isinstance(rinexfn, Path)
+
+    if info["rinextype"] == "sp3":
+        return load_sp3(rinexfn, outfn)
     elif rinexfn.suffix == ".nc":
         # outfn not used here, because we already have the converted file!
         try:
@@ -112,10 +115,10 @@ def batch_convert(
     path: Path,
     glob: str,
     out: Path,
-    use: Sequence[str] = None,
-    tlim: Tuple[datetime, datetime] = None,
+    use: list[str] = None,
+    tlim: tuple[datetime, datetime] = None,
     useindicators: bool = False,
-    meas: Sequence[str] = None,
+    meas: list[str] = None,
     verbose: bool = False,
     *,
     fast: bool = True,
@@ -142,15 +145,15 @@ def batch_convert(
 
 
 def rinexnav(
-    fn: Union[TextIO, str, Path],
+    fn: T.TextIO | str | Path,
     outfn: Path = None,
-    use: Sequence[str] = None,
+    use: list[str] = None,
     group: str = "NAV",
-    tlim: Tuple[datetime, datetime] = None,
+    tlim: tuple[datetime, datetime] = None,
     *,
     overwrite: bool = False,
 ) -> xarray.Dataset:
-    """ Read RINEX 2 or 3  NAV files"""
+    """Read RINEX 2 or 3  NAV files"""
 
     if isinstance(fn, (str, Path)):
         fn = Path(fn).expanduser()
@@ -186,18 +189,18 @@ def rinexnav(
 
 
 def rinexobs(
-    fn: Union[TextIO, str, Path],
+    fn: T.TextIO | Path,
     outfn: Path = None,
-    use: Sequence[str] = None,
+    use: list[str] = None,
     group: str = "OBS",
-    tlim: Tuple[datetime, datetime] = None,
+    tlim: tuple[datetime, datetime] = None,
     useindicators: bool = False,
-    meas: Sequence[str] = None,
+    meas: list[str] = None,
     verbose: bool = False,
     *,
     overwrite: bool = False,
     fast: bool = True,
-    interval: Union[float, int, timedelta] = None,
+    interval: float | int | timedelta = None,
 ) -> xarray.Dataset:
     """
     Read RINEX 2.x and 3.x OBS files in ASCII or GZIP (or Hatanaka)

src/georinex/common.py

@@ -1,6 +1,7 @@
+from __future__ import annotations
+import typing as T
 from pathlib import Path
 from datetime import timedelta
-from typing import Dict, Union, Any
 import numpy as np
 import logging
 
@@ -26,20 +27,23 @@ def rinex_string_to_float(s: str) -> float:
     return float(s.replace("D", "E"))
 
 
-def check_ram(memneed: int, fn: Path):
+def check_ram(memneed: int, fn: T.TextIO | Path):
     if psutil is None:
         return
 
     mem = psutil.virtual_memory()
 
     if memneed > 0.5 * mem.available:  # because of array copy Numpy => Xarray
-        raise RuntimeError(
-            f"{fn} needs {memneed/1e9} GBytes RAM, but only {mem.available/1e9} Gbytes available \n"
+        errmsg = (
+            f"needs {memneed/1e9} GBytes RAM, but only {mem.available/1e9} Gbytes available \n"
             "try fast=False to reduce RAM usage, raise a GitHub Issue to let us help"
         )
+        if isinstance(fn, Path):
+            errmsg = f"{fn}" + errmsg
+        raise RuntimeError(errmsg)
 
 
-def determine_time_system(header: Dict[str, Any]) -> str:
+def determine_time_system(header: dict[str, T.Any]) -> str:
     """Determine which time system is used in an observation file."""
     # Current implementation is quite inconsistent in terms what is put into
     # header.
@@ -70,7 +74,7 @@ def determine_time_system(header: Dict[str, Any]) -> str:
     return ts
 
 
-def check_time_interval(interval: Union[float, int, timedelta]) -> timedelta:
+def check_time_interval(interval: float | int | timedelta) -> timedelta:
     if isinstance(interval, (float, int)):
         if interval < 0:
             raise ValueError("time interval must be non-negative")

src/georinex/geo.py

@@ -1,14 +1,13 @@
+from __future__ import annotations
 import pandas
-from typing.io import TextIO
-from typing import Sequence, Union
 import io
 import xarray
 from pathlib import Path
 
 from .utils import rinexheader
 
 
-def get_locations(files: Union[TextIO, Sequence[Path]]) -> pandas.DataFrame:
+def get_locations(files: list[Path]) -> pandas.DataFrame:
     """
     retrieve locations of GNSS receivers
 
@@ -19,21 +18,23 @@ def get_locations(files: Union[TextIO, Sequence[Path]]) -> pandas.DataFrame:
 
     if isinstance(files[0], io.StringIO):
         locs = pandas.DataFrame(index=["0"], columns=["lat", "lon", "interval"])
+    elif isinstance(files[0], Path):
+        locs = pandas.DataFrame(index=[file.name for file in files], columns=["lat", "lon", "interval"])
     else:
-        locs = pandas.DataFrame(index=[f.name for f in files], columns=["lat", "lon", "interval"])
+        raise TypeError("Expecting pathlib.Path")
 
-    for f in files:
-        if isinstance(f, Path) and f.suffix == ".nc":
-            dat = xarray.open_dataset(f, group="OBS")
+    for file in files:
+        if isinstance(file, Path) and file.suffix == ".nc":
+            dat = xarray.open_dataset(file, group="OBS")
             hdr = dat.attrs
         else:
             try:
-                hdr = rinexheader(f)
+                hdr = rinexheader(file)
             except ValueError:
                 continue
 
-        if isinstance(f, Path):
-            key = f.name
+        if isinstance(file, Path):
+            key = file.name
         else:
             key = "0"
 

src/georinex/hatanaka.py

@@ -4,10 +4,11 @@
 NOTE: This is a candidate for importlib.resources in Python >= 3.7
 """
 
+from __future__ import annotations
+import typing as T
 import subprocess
 import shutil
 from pathlib import Path
-from typing.io import TextIO
 
 from .build import build
 
@@ -45,7 +46,7 @@ def crxexe(path: Path = Path(__file__).parent / "rnxcmp") -> str:
         raise RuntimeError("Hatanaka converter is broken.")
 
 
-def opencrx(f: TextIO) -> str:
+def opencrx(f: T.TextIO) -> str:
     """
     Conversion to string is necessary because of a quirk where gzip.open() even with 'rt' doesn't decompress until read.
 

src/georinex/keplerian.py

@@ -2,13 +2,14 @@
 GPS Keplerian elements => ECEF
 Michael Hirsch, Ph.D.
 """
+
+from __future__ import annotations
 from datetime import datetime, timedelta
 import xarray
 import numpy as np
-from typing import Tuple
 
 
-def keplerian2ecef(sv: xarray.DataArray) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
+def keplerian2ecef(sv: xarray.DataArray) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
     """
     based on:
     https://ascelibrary.org/doi/pdf/10.1061/9780784411506.ap03

src/georinex/nav2.py

@@ -1,8 +1,7 @@
-#!/usr/bin/env python
+from __future__ import annotations
+import typing as T
 from pathlib import Path
 from datetime import datetime
-from typing import Dict, Union, Any, Sequence
-from typing.io import TextIO
 import xarray
 import numpy as np
 import logging
@@ -15,7 +14,7 @@
 Nl = {"G": 7, "R": 3, "E": 7}  # number of additional SV lines
 
 
-def rinexnav2(fn: Union[TextIO, str, Path], tlim: Sequence[datetime] = None) -> xarray.Dataset:
+def rinexnav2(fn: T.TextIO | Path, tlim: tuple[datetime, datetime] = None) -> xarray.Dataset:
     """
     Reads RINEX 2.x NAV files
     Michael Hirsch, Ph.D.
@@ -214,7 +213,7 @@ def rinexnav2(fn: Union[TextIO, str, Path], tlim: Sequence[datetime] = None) ->
     return nav
 
 
-def navheader2(f: TextIO) -> Dict[str, Any]:
+def navheader2(f: T.TextIO) -> dict[str, T.Any]:
     """
     For RINEX NAV version 2 only. End users should use rinexheader()
     """
@@ -254,12 +253,12 @@ def _timenav(ln: str) -> datetime:
     )
 
 
-def _skip(f: TextIO, Nl: int):
+def _skip(f: T.TextIO, Nl: int):
     for _, _ in zip(range(Nl), f):
         pass
 
 
-def navtime2(fn: Union[TextIO, Path]) -> np.ndarray:
+def navtime2(fn: T.TextIO | Path) -> np.ndarray:
     """
     read all times in RINEX 2 NAV file
     """

src/georinex/nav3.py

@@ -1,12 +1,12 @@
-#!/usr/bin/env python
+from __future__ import annotations
+import typing as T
 from pathlib import Path
 import xarray
 import logging
 import numpy as np
 import math
 from datetime import datetime
-from typing import Dict, Union, List, Any, Sequence
-from typing.io import TextIO
+
 
 #
 from .rio import opener, rinexinfo
@@ -19,7 +19,7 @@
 
 
 def rinexnav3(
-    fn: Union[TextIO, str, Path], use: Sequence[str] = None, tlim: Sequence[datetime] = None
+    fn: T.TextIO | Path, use: list[str] = None, tlim: tuple[datetime, datetime] = None
 ) -> xarray.Dataset:
     """
     Reads RINEX 3.x NAV files
@@ -34,9 +34,9 @@ def rinexnav3(
 
     svs = []
     raws = []
-    svtypes: List[str] = []
-    fields: Dict[str, List[str]] = {}
-    times: List[datetime] = []
+    svtypes: list[str] = []
+    fields: dict[str, list[str]] = {}
+    times: list[datetime] = []
 
     with opener(fn) as f:
         header = navheader3(f)
@@ -166,7 +166,7 @@ def rinexnav3(
     return nav
 
 
-def _skip(f: TextIO, Nl: int):
+def _skip(f: T.TextIO, Nl: int):
     for _, _ in zip(range(Nl), f):
         pass
 
@@ -183,7 +183,7 @@ def _time(ln: str) -> datetime:
     )
 
 
-def _sparefields(cf: List[str], sys: str, raw: str) -> List[str]:
+def _sparefields(cf: list[str], sys: str, raw: str) -> list[str]:
     """
     check for optional spare fields, or GPS "fit interval" field
 
@@ -228,7 +228,7 @@ def _sparefields(cf: List[str], sys: str, raw: str) -> List[str]:
     return cf
 
 
-def _newnav(ln: str, sv: str) -> List[str]:
+def _newnav(ln: str, sv: str) -> list[str]:
 
     if sv.startswith("G"):
         """
@@ -461,7 +461,7 @@ def _newnav(ln: str, sv: str) -> List[str]:
     return fields
 
 
-def navheader3(f: TextIO) -> Dict[str, Any]:
+def navheader3(f: T.TextIO) -> dict[str, T.Any]:
 
     if isinstance(f, (str, Path)):
         with opener(f, header=True) as h:
@@ -491,7 +491,7 @@ def navheader3(f: TextIO) -> Dict[str, Any]:
     return hdr
 
 
-def navtime3(fn: Union[TextIO, Path]) -> np.ndarray:
+def navtime3(fn: T.TextIO | Path) -> np.ndarray:
     """
     return all times in RINEX file
     """