Merge pull request #99 from JohnMarzulli/dynon_reconnect

Cleanup and Simplify Data Acquisition.

data_sources/ahrs_simulation.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 from common_utils import fast_math, geo_math, simulated_values
 
@@ -16,7 +16,8 @@ def simulate(
         """
         Ticks the simulated data.
         """
-        hours_since_start = ((datetime.utcnow() - self.__starting_time__).total_seconds() / 60.0) / 60.0
+        hours_since_start = (datetime.now(timezone.utc) - self.__starting_time__).total_seconds() / 60.0 / 60.0
+
         proportion_into_simulation = hours_since_start / self.__hours_to_destination__
 
         simulated_lat = fast_math.interpolatef(self.__starting_lat__, self.__ending_lat__, proportion_into_simulation)
@@ -67,7 +68,7 @@ def __init__(
         self.ahrs_data.position = [self.__starting_lat__, self.__starting_long__]
 
         self.__hours_to_destination__ = 17.5
-        self.__starting_time__ = datetime.utcnow()
+        self.__starting_time__ = datetime.now(timezone.utc)
         self.__end_time__ = self.__starting_time__ + timedelta(hours=17.5)
 
         self.pitch_simulator = simulated_values.SimulatedValue(1, 30, -1)

data_sources/aircraft.py

@@ -36,7 +36,7 @@ def update_orientation_in_background(
             except KeyboardInterrupt:
                 raise
             except Exception as ex:
-                self.warn("update_orientation_in_background ex={}".format(ex))
+                self.warn(f"update_orientation_in_background ex={ex}")
 
     def __init__(
         self,
@@ -94,5 +94,5 @@ def __update_avionics_orientation__(
     plane = Aircraft()
 
     while True:
-        print(str(plane.get_orientation().roll))
+        print(plane.get_orientation().roll)
         time.sleep(1.0 / 60.0)

data_sources/gdl90_ahrs_source.py

@@ -2,7 +2,7 @@
 Module to extract GDL-90 data and turn it into a common, usable form.
 """
 
-from datetime import datetime
+from datetime import datetime, timezone
 
 import requests
 from common_utils import data_cache, logging_object
@@ -21,26 +21,6 @@ class AhrsStratux(logging_object.LoggingObject):
     Class to pull actual AHRS data from a Stratux (or Stratus)
     """
 
-    def __get_value__(
-        self,
-        ahrs_json: dict,
-        key: str,
-        default
-    ):
-        """
-        Safely return the value from the AHRS blob
-
-        Args:
-            ahrs_json (dict): The AHRS data to extract the potential data from.
-            key (str): The name of the data we want.
-            default: The default value if the data is not found.
-        """
-
-        if key in ahrs_json:
-            return ahrs_json[key]
-
-        return default
-
     def __get_value_with_fallback__(
         self,
         ahrs_json: dict,
@@ -50,10 +30,10 @@ def __get_value_with_fallback__(
         if keys is None:
             return default
 
-        values = [self.__get_value__(ahrs_json, key, default) for key in keys]
+        values = [ahrs_json.get(key, default) for key in keys]
         values = list(filter(lambda x: x != default, values))
 
-        return values[0] if values is not None and len(values) > 0 else default
+        return values[0] if values is not None and values else default
 
     def __decode_situation__(
         self,
@@ -73,40 +53,33 @@ def __decode_situation__(
         """
         new_ahrs_data = ahrs_data.AhrsData()
 
-        system_utc_time = str(datetime.utcnow())
+        system_utc_time = str(datetime.now(timezone.utc))
 
         new_ahrs_data.is_avionics_source = ahrs_json is not None\
             and "Service" in ahrs_json\
             and "ToHud" in ahrs_json["Service"]\
             and len(ahrs_json) > 1
 
-        new_ahrs_data.gps_online = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.gps_online = ahrs_json.get(
             'GPSFixQuality',
             0) > 0
 
-        latitude = self.__get_value__(
-            ahrs_json,
+        latitude = ahrs_json.get(
             'GPSLatitude',
             None) if new_ahrs_data.gps_online else None
-        longitude = self.__get_value__(
-            ahrs_json,
+        longitude = ahrs_json.get(
             'GPSLongitude',
             None) if new_ahrs_data.gps_online else None
-        new_ahrs_data.roll = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.roll = ahrs_json.get(
             'AHRSRoll',
             0.0)
-        new_ahrs_data.pitch = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.pitch = ahrs_json.get(
             'AHRSPitch',
             0.0)
-        new_ahrs_data.compass_heading = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.compass_heading = ahrs_json.get(
             'AHRSGyroHeading',
             1080)  # anything above 360 indicates "not available"
-        new_ahrs_data.gps_heading = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.gps_heading = ahrs_json.get(
             'GPSTrueCourse',
             ahrs_data.NOT_AVAILABLE)
         new_ahrs_data.alt = self.__get_value_with_fallback__(
@@ -118,25 +91,20 @@ def __decode_situation__(
             ahrs_json,
             ["BaroVerticalSpeed", 'GPSVerticalSpeed'],
             ahrs_data.NOT_AVAILABLE)
-        new_ahrs_data.airspeed = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.airspeed = ahrs_json.get(
             'AHRSAirspeed',
             ahrs_data.NOT_AVAILABLE)
-        new_ahrs_data.groundspeed = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.groundspeed = ahrs_json.get(
             'GPSGroundSpeed',
             ahrs_data.NOT_AVAILABLE) if new_ahrs_data.gps_online else ahrs_data.NOT_AVAILABLE
-        new_ahrs_data.g_load = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.g_load = ahrs_json.get(
             'AHRSGLoad',
             ahrs_data.NOT_AVAILABLE)
-        new_ahrs_data.utc_time = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.utc_time = ahrs_json.get(
             'GPSTime',
             system_utc_time)
 
-        new_ahrs_data.slip_skid = self.__get_value__(
-            ahrs_json,
+        new_ahrs_data.slip_skid = ahrs_json.get(
             'AHRSSlipSkid',
             None
         )

data_sources/receiver_capabilities.py

@@ -24,12 +24,10 @@ def __get_value__(
         if key in self.__capabilities_json__:
             try:
                 return self.__capabilities_json__[key]
-            except KeyboardInterrupt:
-                raise
-            except SystemExit:
+            except (KeyboardInterrupt, SystemExit):
                 raise
             except Exception as ex:
-                self.warn("__get_capability__ EX={}".format(ex))
+                self.warn(f"__get_capability__ EX={ex}")
                 return None
 
         return None
@@ -50,14 +48,11 @@ def __get_capability__(
         if key in self.__capabilities_json__:
             try:
                 return bool(self.__capabilities_json__[key])
-            except KeyboardInterrupt:
-                raise
-            except SystemExit:
+            except (KeyboardInterrupt, SystemExit):
                 raise
             except Exception as ex:
-                self.warn("__get_capability__ EX={}".format(ex))
+                self.warn(f"__get_capability__ EX={ex}")
                 return False
-
         return False
 
     def __init__(
@@ -87,14 +82,11 @@ def __init__(
                 self.__capabilities_json__ = stratux_session.get(
                     url, timeout=2).json()
 
-            except KeyboardInterrupt:
-                raise
-            except SystemExit:
+            except (KeyboardInterrupt, SystemExit):
                 raise
             except Exception as ex:
                 self.__capabilities_json__ = {}
-                self.warn("EX in __init__ ex={}".format(ex))
-
+                self.warn(f"EX in __init__ ex={ex}")
             self.traffic_enabled = self.__get_capability__('UAT_Enabled')
             self.gps_enabled = self.__get_capability__('GPS_Enabled')
             self.barometric_enabled = self.__get_capability__(
@@ -106,7 +98,7 @@ def __init__(
                 # Ownship is in Hex... traffic reports come in int...
                 self.ownship_icao = int(
                     self.ownship_mode_s, 16) if self.ownship_mode_s is not None else 0
-            except:
+            except Exception:
                 self.ownship_icao = 0
             # http://192.168.10.1/getSettings - get device settings. Example output:
             #

data_sources/receiver_status.py

@@ -20,12 +20,10 @@ def __get_status__(
         if key in self.__status_json__:
             try:
                 return bool(self.__status_json__[key])
-            except KeyboardInterrupt:
-                raise
-            except SystemExit:
+            except (KeyboardInterrupt, SystemExit):
                 raise
             except Exception as ex:
-                self.warn("__get_status__ EX={}".format(ex))
+                self.warn(f"__get_status__ EX={ex}")
                 return False
 
         return False
@@ -55,14 +53,11 @@ def __init__(
                 self.__status_json__ = stratux_session.get(
                     url, timeout=2).json()
 
-            except KeyboardInterrupt:
-                raise
-            except SystemExit:
+            except (KeyboardInterrupt, SystemExit):
                 raise
             except Exception as ex:
-                self.warn("__get_status__ EX={}".format(ex))
+                self.warn(f"__get_status__ EX={ex}")
                 self.__status_json__ = {}
-
             self.cpu_temp = self.__get_status__('CPUTemp')
             self.satellites_locked = self.__get_status__(
                 'GPS_satellites_locked')

data_sources/targets.py

@@ -2,6 +2,7 @@
 Manages heading/target bugs.
 """
 
+import contextlib
 import json
 
 import configuration
@@ -21,7 +22,7 @@ def save(
                 json_config_file.write(json_config_text)
 
             return True
-        except:
+        except Exception:
             return False
 
     def clear_targets(
@@ -60,7 +61,7 @@ def __init__(
 
         self.targets = []
 
-        try:
+        with contextlib.suppress(Exception):
             with open(configuration.HEADING_BUGS_FILE) as json_config_file:
                 json_config_text = json_config_file.read()
                 json_config = json.loads(json_config_text)
@@ -70,8 +71,6 @@ def __init__(
                         if len(lat_long_altitude) == 3:
                             self.add_target(
                                 lat_long_altitude[0], lat_long_altitude[1], lat_long_altitude[2])
-        except:
-            pass
 
 
 TARGET_MANAGER = Targets()