Fix malformed template, cache CelesTrak TLEs

README.md

@@ -1,9 +1,9 @@
 # UniClOGS Pass Commander
 This software controls the local functions of a
-[UniClOGS](https://www.oresat.org/technologies/ground-stations) for sending
-commands to the [OreSat0](https://www.oresat.org/satellites/oresat0) and
-[OreSat0.5](https://www.oresat.org/satellites/oresat0-5)
-[CubeSats](https://en.wikipedia.org/wiki/CubeSat).
+[UniClOGS](https://www.oresat.org/technologies/ground-stations) ground station
+for sending commands to [CubeSats](https://en.wikipedia.org/wiki/CubeSat), as
+used with the [OreSat0](https://www.oresat.org/satellites/oresat0)
+and [OreSat0.5](https://www.oresat.org/satellites/oresat0-5) missions.
 
 ## Major functions
 * Tracks satellites using the excellent [Skyfield](https://rhodesmill.org/skyfield/)
@@ -22,6 +22,8 @@ commands to the [OreSat0](https://www.oresat.org/satellites/oresat0) and
   to manage Doppler shifting and to send command packets
 
 ## Installing
+Requires Linux with Python 3.11 or greater.
+
 ```sh
 git clone https://github.com/uniclogs/uniclogs-pass-commander.git
 sudo apt install python3-pip
@@ -32,9 +34,17 @@ Running `pass-commander --template` will generate a
 template configuration file. You should receive instructions for editing it. Go
 do that now (see below for detailed description).
 
-When your config is all set up, run with `pass-commander`. See the
-`--help` flag for more options. For example `pass-commander -s 60525
--m all -a dryrun`.
+When your config is all set up, run with `pass-commander`. See the `--help` flag
+for more options. Initially you'll not have any saved TLEs so either find one
+for your satellite of interest and add it to `TleCache` in `pass_commander.toml`
+or run without the `--mock tle` flag to download one locally:
+```sh
+pass-commander --satellite 60525 --action dryrun -m tx -m con -m rot
+```
+After that the `--mock all` flag can be used for brevity:
+```sh
+pass-commander --satellite 60525 --action dryrun --mock all
+```
 
 Testing without rotctld, stationd and a running radio flowgraph is partially
 supported. See the `--mock` flag, especially `-m all`.

pass_commander/commander.py

@@ -339,7 +339,10 @@ def sleep_until_next_pass(self) -> tuple[Satellite, PassInfo]:
 
         while True:
             sat = Satellite(
-                self.conf.sat_id, tle_cache=self.conf.tle_cache, local_only='con' in self.conf.mock
+                self.conf.sat_id,
+                self.conf.dir,
+                tle_cache=self.conf.tle_cache,
+                local_only='tle' in self.conf.mock,
             )
             np = self.track.next_pass(sat, min_el=self.conf.min_el)
             if np is None:
@@ -418,7 +421,10 @@ def point(self, coord: GeographicPosition) -> None:
 
     def dryrun(self) -> None:
         sat = Satellite(
-            self.conf.sat_id, tle_cache=self.conf.tle_cache, local_only='con' in self.conf.mock
+            self.conf.sat_id,
+            self.conf.dir,
+            tle_cache=self.conf.tle_cache,
+            local_only='tle' in self.conf.mock,
         )
         np = self.track.next_pass(sat, min_el=Angle(degrees=80), lookahead=timedelta(days=30))
         if np is None:

pass_commander/config.py

@@ -169,6 +169,8 @@ class Config:
     # Command line only
     mock: set[str] = field(default_factory=set)
     pass_count: int = 9999
+    # FIXME: use XDG_CONFIG_HOME
+    dir: PosixPath = PosixPath('~/.config/OreSat').expanduser()  # noqa: RUF009
 
     def __post_init__(self, path: Path) -> None:
         '''Load a config from a given file.
@@ -188,8 +190,11 @@ def __post_init__(self, path: Path) -> None:
         path
             Path to the config file, usually pass_commander.toml
         '''
+        path = path.expanduser()
+        self.dir = PosixPath(path.parent)
+
         try:
-            config = tomlkit.parse(path.expanduser().read_text())
+            config = tomlkit.parse(path.read_text())
         except tomlkit.exceptions.ParseError as e:
             raise InvalidTomlError(*e.args) from e
         except FileNotFoundError as e:
@@ -259,7 +264,7 @@ def template(cls, path: Path) -> None:
 
         hosts = tomlkit.table()
         hosts['radio'] = str(cls.flowgraph[0])
-        hosts['station'] = str(cls.station)
+        hosts['station'] = str(cls.station[0])
         hosts['rotator'] = str(cls.rotator)
 
         observer = tomlkit.table()

pass_commander/main.py

@@ -30,7 +30,7 @@ def handle_args() -> Namespace:  # noqa: D103
     parser.add_argument(
         "-c",
         "--config",
-        default="~/.config/OreSat/pass_commander.toml",
+        default=config.Config.dir / "pass_commander.toml",
         type=Path,
         help=dedent(
             """\
@@ -54,13 +54,14 @@ def handle_args() -> Namespace:  # noqa: D103
         "-m",
         "--mock",
         action="append",
-        choices=("tx", "rot", "con", "all"),
+        choices=("tx", "rot", "con", "tle", "all"),
         help=dedent(
             """\
             Use a simulated (mocked) external dependency, not the real thing
             - tx: No PTT or EDL bytes sent to flowgraph
             - rot: No actual movement commanded for the rotator
-            - con: Don't use network services - TLEs, weather, rot2prog, stationd
+            - con: Don't use network services - weather, rot2prog, stationd
+            - tle: Only use locally saved TLEs, don't fetch from the internet (CelesTrak)
             - all: All of the above
             Can be issued multiple times, e.g. '-m tx -m rot' will disable tx and rotator"""
         ),
@@ -152,7 +153,7 @@ def main() -> None:  # noqa: D103 C901 PLR0912 PLR0915
     else:
         conf.mock = set(args.mock or [])
         if 'all' in conf.mock:
-            conf.mock = {'tx', 'rot', 'con'}
+            conf.mock = {'tx', 'rot', 'con', 'tle'}
         # Favor command line values over config file values
         conf.txgain = args.tx_gain or conf.txgain
         conf.sat_id = args.satellite or conf.sat_id

pass_commander/satellite.py

@@ -1,6 +1,7 @@
 import logging
 import re
 from pathlib import Path
+from time import time
 
 import requests
 from skyfield.api import EarthSatellite
@@ -12,7 +13,12 @@
 
 class Satellite(EarthSatellite):  # type: ignore[misc]
     def __init__(
-        self, sat_id: str, *, tle_cache: TleCache | None = None, local_only: bool = False
+        self,
+        sat_id: str,
+        conf_dir: Path,
+        *,
+        tle_cache: TleCache | None = None,
+        local_only: bool = False,
     ) -> None:
         '''Fetch a TLE and build a satellite.
 
@@ -32,6 +38,9 @@ def __init__(
         sat_id
             The ID of the satellite, either as an International Designator, a NORAD Satellite
             Catalog number, or Catalog satellite name.
+        conf_dir
+            Path to the directory to save TLEs from Celestrak in. When in doubt the same directory
+            that pass_commander.toml is in is good.
         local_only
             If true, do not use the internet for TLE lookup, only tle_cache or Gpredict.
         tle_cache
@@ -42,25 +51,37 @@ def __init__(
             query = "INTDES"
         elif re.match(r"^\d{1,9}$", sat_id):
             query = "CATNR"
+        filename = conf_dir / f'{query}-{sat_id}.txt'
 
-        tle = None
-        if tle is None and not local_only:
-            # see https://celestrak.org/NORAD/documentation/gp-data-formats.php
-            # FIXME: 2 hour cache. This must add the data to the TLE cache. Just use skytraq?
+        # see https://celestrak.org/NORAD/documentation/gp-data-formats.php
+        # In particular a 2 hour minimum cache time. In practice I think we get one new TLE per day
+        # so here we wait 0.5 days. FIXME: Add to TLE cache/save in pass_commander.toml? Depends on
+        # pydantic.
+        expired = not filename.exists() or (time() - filename.stat().st_mtime > 12 * 60 * 60)
+        if not local_only and expired:
+            logger.info("fetching TLE from celestrak for %s", sat_id)
             r = requests.get(
                 "https://celestrak.org/NORAD/elements/gp.php",
                 params={query: sat_id},
                 timeout=10,
             )
             r.raise_for_status()
-            lines = r.text.splitlines()
+            filename.write_text(r.text)
+
+        tle = None
+        if tle is None and filename.exists():
+            logger.info("using cached TLE from %s", filename)
+            lines = filename.read_text().splitlines()
+            # Not documented but CelesTrak currently returns this string if the previous query
+            # didn't match an object it knows about. We have to cache it anyway to respect their
+            # policy.
             if lines[0] == "No GP data found":
                 logger.info("No results for %s at celestrak", sat_id)
             else:
                 tle = lines
 
         if tle is None and tle_cache and sat_id in tle_cache:
-            logger.info("using cached TLE")
+            logger.info("using cached TLE from pass_commander.toml")
             tle = tle_cache[sat_id]
 
         if tle is None and query == "CATNR":

tests/conftest.py

@@ -119,6 +119,6 @@ def edl() -> tuple[str, int]:
 
 
 @pytest.fixture(params=tles)
-def sat(request) -> Satellite:  # noqa: ANN001
+def sat(request, tmp_path: Path) -> Satellite:  # noqa: ANN001
     name = request.param[0]
-    return Satellite(name, tle_cache={name: request.param}, local_only=True)
+    return Satellite(name, tmp_path, tle_cache={name: request.param}, local_only=True)

tests/test_config.py

@@ -165,9 +165,22 @@ def test_edl(self, tmp_path: Path, good_toml: TOMLDocument, edlport: str | float
             Config(path)
 
     def test_template(self, tmp_path: Path) -> None:
-        Config.template(tmp_path / "faketemplate.toml")
+        path = tmp_path / "faketemplate.toml"
+        Config.template(path)
+
+        # Trying to load the template should fail because template text hasn't been removed
         with pytest.raises(config.TemplateTextError):
-            Config(tmp_path / "faketemplate.toml")
+            Config(path)
+
+        # But if we remove the template text
+        with path.open('r') as f:
+            contents = f.read()
+        contents = contents.replace('<', '')
+        contents = contents.replace('>', '')
+        with path.open('w') as f:
+            f.write(contents)
+        # then it should load
+        Config(path)
 
     def test_template_exists(self, tmp_path: Path) -> None:
         conf = tmp_path / "config.toml"

tests/test_satellite.py

@@ -1,3 +1,4 @@
+from pathlib import Path
 from typing import Final
 
 import pytest
@@ -32,17 +33,17 @@ class TestSatellite:
     ]
 
     @responses.activate
-    def test_cache(self) -> None:
+    def test_cache(self, tmp_path: Path) -> None:
         cache = dict.fromkeys(self.names, self.tle)
 
         for name in self.names:
-            Satellite(name, tle_cache=cache, local_only=True)
+            Satellite(name, tmp_path, tle_cache=cache, local_only=True)
 
         with pytest.raises(ValueError, match=r"^Invalid satellite identifier"):
-            Satellite('invalid', tle_cache=cache, local_only=True)
+            Satellite('invalid', tmp_path, tle_cache=cache, local_only=True)
 
     @responses.activate
-    def test_celestrak(self) -> None:
+    def test_celestrak(self, tmp_path: Path) -> None:
         # valid
         # missing
         # timeout
@@ -67,21 +68,21 @@ def test_celestrak(self) -> None:
         responses.add(missing)
         responses.add(forbidden)
 
-        Satellite(self.names[1])
+        Satellite(self.names[1], tmp_path)
 
         with pytest.raises(ValueError, match=r"^Invalid satellite identifier"):
-            Satellite('missing')
+            Satellite('missing', tmp_path)
 
         with pytest.raises(requests.exceptions.HTTPError):
-            Satellite('forbidden')
+            Satellite('forbidden', tmp_path)
 
     @responses.activate
-    def test_cache_fallback(self) -> None:
+    def test_cache_fallback(self, tmp_path: Path) -> None:
         fallback = responses.Response(
             method="GET",
             url="https://celestrak.org/NORAD/elements/gp.php?NAME=fallback",
             body='No GP data found',
         )
         responses.add(fallback)
 
-        Satellite('fallback', tle_cache={'fallback': self.tle})
+        Satellite('fallback', tmp_path, tle_cache={'fallback': self.tle})