Merge pull request #205 from pnuu/fallback_empty_tles_env

Add fallback when `TLES` environment variable not listing any files

Author: mraspaud

pyorbital/tests/test_tlefile.py

@@ -384,6 +384,25 @@ def test_get_uris_and_open_func_using_tles_env(caplog, fake_local_tles_dir, monk
     assert log_message in caplog.text
 
 
+def test_get_uris_and_open_func_using_empty_tles_env(caplog, fake_local_tles_dir, monkeypatch):
+    """Test getting the uris and associated open-function for reading tles.
+
+    Test providing no tle file but using TLES env that returns no files.
+    """
+    from pyorbital.tlefile import _get_uris_and_open_func
+
+    monkeypatch.setenv("TLES", "/no/files/here/tle*txt")
+
+    with caplog.at_level(logging.DEBUG):
+        with pytest.warns():
+            uris, _ = _get_uris_and_open_func()
+
+    warning_text = "TLES environment variable points to no TLE files"
+    debug_text = "Fetch TLE from the internet."
+    assert warning_text in caplog.text
+    assert debug_text in caplog.text
+
+
 class TLETest(unittest.TestCase):
     """Test TLE reading.
 

pyorbital/tlefile.py

@@ -24,6 +24,7 @@
 import logging
 import os
 import sqlite3
+import warnings
 from itertools import zip_longest
 from urllib.request import urlopen
 
@@ -294,36 +295,58 @@ def _get_local_tle_path_from_env():
 
 def _get_uris_and_open_func(tle_file=None):
     """Get the uri's and the adequate file open call for the TLE files."""
-    def _open(filename):
-        return io.open(filename, "rb")
-
     local_tle_path = _get_local_tle_path_from_env()
 
     if tle_file:
-        if isinstance(tle_file, io.StringIO):
-            uris = (tle_file,)
-            open_func = _dummy_open_stringio
-        elif "ADMIN_MESSAGE" in tle_file:
-            uris = (io.StringIO(read_tle_from_mmam_xml_file(tle_file)),)
-            open_func = _dummy_open_stringio
-        else:
-            uris = (tle_file,)
-            open_func = _open
+        uris, open_func = _get_tle_file_uris_and_open_method(tle_file)
     elif local_tle_path:
-        # TODO: get the TLE file closest in time to the actual satellite
-        # overpass, NOT the latest!
-        list_of_tle_files = glob.glob(local_tle_path)
+        uris, open_func = _get_local_uris_and_open_method(local_tle_path)
+    else:
+        uris, open_func = _get_internet_uris_and_open_method()
+    return uris, open_func
+
+
+def _get_tle_file_uris_and_open_method(tle_file):
+    if isinstance(tle_file, io.StringIO):
+        uris = (tle_file,)
+        open_func = _dummy_open_stringio
+    elif "ADMIN_MESSAGE" in tle_file:
+        uris = (io.StringIO(read_tle_from_mmam_xml_file(tle_file)),)
+        open_func = _dummy_open_stringio
+    else:
+        uris = (tle_file,)
+        open_func = _open
+    return uris, open_func
+
+
+def _open(filename):
+    return io.open(filename, "rb")
+
+
+def _get_local_uris_and_open_method(local_tle_path):
+    # TODO: get the TLE file closest in time to the actual satellite
+    # overpass, NOT the latest!
+    list_of_tle_files = glob.glob(local_tle_path)
+    if list_of_tle_files:
         uris = (max(list_of_tle_files, key=os.path.getctime), )
         LOGGER.debug("Reading TLE from %s", uris[0])
         open_func = _open
     else:
-        LOGGER.debug("Fetch TLE from the internet.")
-        uris = TLE_URLS
-        open_func = urlopen
+        LOGGER.warning("TLES environment variable points to no TLE files")
+        throttle_warning = "TLEs will be downloaded from Celestrak, which can throttle the connection."
+        LOGGER.warning(throttle_warning)
+        warnings.warn(throttle_warning)
+
+        uris, open_func = _get_internet_uris_and_open_method()
 
     return uris, open_func
 
 
+def _get_internet_uris_and_open_method():
+    LOGGER.debug("Fetch TLE from the internet.")
+    return TLE_URLS, urlopen
+
+
 def _get_first_tle(uris, open_func, platform=""):
     return _get_tles_from_uris(uris, open_func, platform=platform, only_first=True)