provide http_get via requests.

.github/workflows/ci-test.yml

@@ -50,6 +50,7 @@ jobs:
           sudo ldconfig -v && \
           poetry config virtualenvs.create false && \
           poetry install --no-interaction && \
+          poetry install --no-interaction -C utils/mavlink-api && \
           sudo dpkg -r --force-depends python3-numpy
       - name: test gamutrf-samples2raw
         env:

gamutrf/mqtt_reporter.py

@@ -5,8 +5,8 @@
 import time
 
 import gpsd
-import httpx
 import paho.mqtt.client as mqtt
+from gamutrf.utils import http_get
 
 
 class MQTTReporter:
@@ -66,23 +66,19 @@
 
     def get_heading(self):
         if self.use_external_heading:
-            try:
-                self.heading = float(
-                    json.loads(
-                        httpx.get(
-                            f"http://{self.external_gps_server}:{self.external_gps_server_port}/heading"
-                        ).text
-                    )["heading"]
-                )
-            except Exception as err:
-                logging.error("could not update external heading: %s", err)
+            heading_result = http_get(
+                f"http://{self.external_gps_server}:{self.external_gps_server_port}/heading"
+            )
+            if heading_result is None:
+                logging.error("could not update external heading")
+            else:
+                self.heading = float(json.loads(heading_result.text)["heading"])
         else:
-            try:
-                self.heading = str(
-                    float(httpx.get(f"http://{self.gps_server}:8000/v1/heading").text)
-                )
-            except Exception as err:
-                logging.error("could not update heading: %s", err)
+            heading_result = http_get(f"http://{self.gps_server}:8000/v1/heading")
+            if heading_result is None:
+                logging.error("could not update heading")
+            else:
+                self.heading = str(float(heading_result.text))
 
     def add_gps(self, publish_args):
         if not self.gps_configured:
@@ -100,13 +96,13 @@
 
         # Use external external GPS
         if self.use_external_gps:
-            try:
-                self.external_gps_msg = json.loads(
-                    httpx.get(
-                        f"http://{self.external_gps_server}:{self.external_gps_server_port}/gps-data"
-                    ).text
-                )
-
+            external_gps_msg = http_get(
+                f"http://{self.external_gps_server}:{self.external_gps_server_port}/gps-data"
+            )
+            if external_gps_msg is None:
+                logging.error("could not update with external GPS")
+            else:
+                self.external_gps_msg = json.loads(external_gps_msg.text)
                 publish_args.update(
                     {
                         "position": (
@@ -121,9 +117,6 @@
                     }
                 )
 
-            except Exception as err:
-                logging.error("could not update with external GPS: %s", err)
-
         # Use internal GPIO GPS
         else:
             try:

gamutrf/sigfinder.py

@@ -11,7 +11,6 @@
 import falcon
 import jinja2
 import numpy as np
-import requests
 import schedule
 
 from prometheus_client import Counter
@@ -25,7 +24,7 @@
 from gamutrf.sigwindows import parse_freq_excluded
 from gamutrf.sigwindows import find_sig_windows
 from gamutrf.sigwindows import ROLLING_FACTOR
-from gamutrf.utils import rotate_file_n, SCAN_FRES
+from gamutrf.utils import http_get, rotate_file_n, SCAN_FRES
 from gamutrf.zmqreceiver import ZmqReceiver, parse_scanners
 
 
@@ -255,14 +254,7 @@
 
 
 def recorder_req(recorder, recorder_args, timeout):
-    url = f"{recorder}/v1/{recorder_args}"
-    try:
-        req = requests.get(url, timeout=timeout)
-        logging.debug(str(req))
-        return req
-    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as err:
-        logging.debug(str(err))
-        return None
+    return http_get(f"{recorder}/v1/{recorder_args}", timeout=timeout)
 
 
 def get_freq_exclusions(args):

gamutrf/utils.py

@@ -1,10 +1,12 @@
 #!/usr/bin/python3
+import logging
 import os
 import re
 import sys
 from pathlib import Path
 
 import numpy as np
+import requests
 
 MTU = 8962
 SCAN_FRES = 1e4
@@ -121,3 +123,13 @@
         for path in Path(filedir).rglob(glob)
         if not os.path.basename(path).startswith(".")
     ]
+
+
+def http_get(url, timeout=10):
+    try:
+        req = requests.get(url, timeout=timeout)
+        logging.debug(str(req))
+        return req
+    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as err:
+        logging.error(f"{url}: {str(err)}")
+        return None