Reformat code

Author: philipbl

UI.py

@@ -14,11 +14,7 @@ def bue_status_table(base_station) -> Table:
     table.add_column("Status", style="yellow", justify="center")
 
     for bue in base_station.connected_bues:
-        status = (
-            "🧪 Testing"
-            if bue in getattr(base_station, "testing_bues", [])
-            else "💤 Idle"
-        )
+        status = "🧪 Testing" if bue in getattr(base_station, "testing_bues", []) else "💤 Idle"
         table.add_row(bUEs[str(bue)], status)
 
     if not base_station.connected_bues:
@@ -51,9 +47,7 @@ def bue_ping_table(base_station) -> Table:
 
 def bue_coordinates_table(base_station) -> Table:
     """Make a styled coordinates table."""
-    table = Table(
-        title="🌍 bUE Coordinates", show_header=True, header_style="bold blue"
-    )
+    table = Table(title="🌍 bUE Coordinates", show_header=True, header_style="bold blue")
     table.add_column("bUE ID", style="cyan", justify="center")
     table.add_column("Coordinates", style="yellow", justify="left")
 
@@ -92,9 +86,7 @@ def bue_distance_table(base_station) -> Table:
                 try:
                     dist = base_station.get_distance(bue1, bue2)
                     if dist is not None:
-                        table.add_row(
-                            f"{bUEs[str(bue1)]} ↔ {bUEs[str(bue2)]}", f"{dist:.2f}m"
-                        )
+                        table.add_row(f"{bUEs[str(bue1)]} ↔ {bUEs[str(bue2)]}", f"{dist:.2f}m")
                     else:
                         table.add_row(
                             f"{bUEs[str(bue1)]} ↔ {bUEs[str(bue2)]}",
@@ -117,9 +109,7 @@ def bue_distance_table(base_station) -> Table:
 
 def received_messages_table(base_station) -> Table:
     """Make a styled coordinates table."""
-    table = Table(
-        title="💌  Received Messages", show_header=True, header_style="bold blue"
-    )
+    table = Table(title="💌  Received Messages", show_header=True, header_style="bold blue")
     table.add_column("Messages", style="cyan", justify="center")
 
     for message in base_station.stdout_history:

base_station_gui.py

@@ -139,9 +139,7 @@ def ping_bue(self, bue_id, lat="", long=""):
 
                 self.bue_timeout_tracker[bue_id] = TIMEOUT + 1
             except Exception as e:
-                logger.bind(bue_id=bue_id).error(
-                    f"ping_bue: Error while handling PING from {bue_id}: {e}"
-                )
+                logger.bind(bue_id=bue_id).error(f"ping_bue: Error while handling PING from {bue_id}: {e}")
 
     """
     # This function will cycle through each bUE the base station should be connected to and make sure that
@@ -161,9 +159,7 @@ def check_bue_timeout(self):
                 logger.bind(bue_id=bue_id).error(f"We missed a PING from {bue_id}")
                 self.bue_timeout_tracker[bue_id] -= 1
             else:
-                logger.error(
-                    f"We haven't heard from {bue_id} in awhile. Maybe disconnected?"
-                )
+                logger.error(f"We haven't heard from {bue_id} in awhile. Maybe disconnected?")
 
     def req_queue_handler(self):
         while not self.EXIT:
@@ -187,48 +183,36 @@ def message_listener(self):
                     bue_id = int(parts[0])
 
                 except Exception as e:
-                    logger.error(
-                        f"message_listener: Failed to parse message '{message}': {e}"
-                    )
+                    logger.error(f"message_listener: Failed to parse message '{message}': {e}")
                     continue  # Skip to the next message
 
                 if "REQ" in message:
                     logger.debug("Sending a CON")
                     self.ota.send_ota_message(bue_id, f"CON:{self.ota.id}")
                     self.bue_timeout_tracker[bue_id] = TIMEOUT
                     if not bue_id in self.connected_bues:
-                        logger.bind(bue_id=bue_id).info(
-                            f"Received a request signal from {bue_id}"
-                        )
+                        logger.bind(bue_id=bue_id).info(f"Received a request signal from {bue_id}")
                         self.connected_bues.append(bue_id)
                     else:
-                        logger.error(
-                            f"Got a connection request from {bue_id} but it is already listed as connected"
-                        )
+                        logger.error(f"Got a connection request from {bue_id} but it is already listed as connected")
 
                 elif "ACK" in message:
                     logger.bind(bue_id=bue_id).info(f"Received ACK from {bue_id}")
 
-                elif (
-                    "PING" in message
-                ):  # Looks like <origin id>,<length>,PING,<lat>,<long>,-55,8
+                elif "PING" in message:  # Looks like <origin id>,<length>,PING,<lat>,<long>,-55,8
                     if len(parts) >= 5:
                         lat = parts[3]
                         long = parts[4]
                     self.ping_bue(bue_id, lat, long)
 
-                elif (
-                    "UPD" in message
-                ):  # 40,55,UPD:LAT,LONG,STDOUT: [helloworld.py STDOUT] TyGoodTest,-42,8
+                elif "UPD" in message:  # 40,55,UPD:LAT,LONG,STDOUT: [helloworld.py STDOUT] TyGoodTest,-42,8
                     if not bue_id in self.testing_bues:
                         self.testing_bues.append(bue_id)
                     lat = parts[3]
                     long = parts[4]
                     stdout = parts[5]
                     # logger.info(f"Received UPD from {bue_id}. Currently at Latitude: {lat}, Longitude: {long}. Message: {stdout}")
-                    logger.bind(bue_id=bue_id).info(
-                        f"Received UPD from {bue_id}. Message: {stdout}"
-                    )
+                    logger.bind(bue_id=bue_id).info(f"Received UPD from {bue_id}. Message: {stdout}")
                     if lat != "" and long != "":
                         self.bue_coordinates[bue_id] = [lat, long]
                     else:
@@ -280,33 +264,23 @@ def get_distance(self, bue_1, bue_2):
                 lat1, lon1 = float(c1[0]), float(c1[1])
                 lat2, lon2 = float(c2[0]), float(c2[1])
             except (ValueError, TypeError):
-                logger.error(
-                    f"Non-numeric coordinates: bUE {bue_1}: {c1}, bUE {bue_2}: {c2}"
-                )
+                logger.error(f"Non-numeric coordinates: bUE {bue_1}: {c1}, bUE {bue_2}: {c2}")
                 return None
 
             # Check if coordinates are within valid ranges
             if not (-90 <= lat1 <= 90) or not (-180 <= lon1 <= 180):
-                logger.error(
-                    f"Invalid latitude/longitude for bUE {bue_1}: lat={lat1}, lon={lon1}"
-                )
+                logger.error(f"Invalid latitude/longitude for bUE {bue_1}: lat={lat1}, lon={lon1}")
                 return None
 
             if not (-90 <= lat2 <= 90) or not (-180 <= lon2 <= 180):
-                logger.error(
-                    f"Invalid latitude/longitude for bUE {bue_2}: lat={lat2}, lon={lon2}"
-                )
+                logger.error(f"Invalid latitude/longitude for bUE {bue_2}: lat={lat2}, lon={lon2}")
                 return None
 
             # Check if coordinates are not empty/zero (optional - depends on your use case)
             if (lat1 == 0 and lon1 == 0) or (lat2 == 0 and lon2 == 0):
-                logger.warning(
-                    f"Zero coordinates detected: bUE {bue_1}: {c1}, bUE {bue_2}: {c2}"
-                )
+                logger.warning(f"Zero coordinates detected: bUE {bue_1}: {c1}, bUE {bue_2}: {c2}")
 
-            logger.info(
-                f"Calculating distance between bUE {bue_1} at ({lat1}, {lon1}) and bUE {bue_2} at ({lat2}, {lon2})"
-            )
+            logger.info(f"Calculating distance between bUE {bue_1} at ({lat1}, {lon1}) and bUE {bue_2} at ({lat2}, {lon2})")
 
             return distance.great_circle((lat1, lon1), (lat2, lon2)).meters
 

base_station_main.py

@@ -120,9 +120,7 @@ def ota_task_queue_handler(self):
     # Send out REQs until a base station is found
     def ota_connect_req(self):
         if self.ota_connected:
-            logger.warning(
-                f"connect_ota_req: OTA device is already connected to base station {self.ota_base_station_id}"
-            )
+            logger.warning(f"connect_ota_req: OTA device is already connected to base station {self.ota_base_station_id}")
             return
 
         # See if there are any new messages from the OTA device
@@ -144,14 +142,10 @@ def ota_connect_req(self):
                     if parts[2].startswith(
                         "CON:"
                     ):  # TY This checks to see if the received message matches a base station connecting
-                        if (
-                            parts[0] == parts[2][4:]
-                        ):  # format. If it does we are now going to be in the connected state
+                        if parts[0] == parts[2][4:]:  # format. If it does we are now going to be in the connected state
                             self.ota_connected = True
                             self.ota_base_station_id = int(parts[0])
-                            logger.info(
-                                f"ota_connect_req: OTA device connected to base station {self.ota_base_station_id}"
-                            )
+                            logger.info(f"ota_connect_req: OTA device connected to base station {self.ota_base_station_id}")
                             self.ota_timeout = TIMEOUT
                             self.ota.send_ota_message(self.ota_base_station_id, "ACK")
                             return
@@ -164,15 +158,11 @@ def ota_connect_req(self):
                 logger.error("ota_connect_req: Error parsing OTA message")
                 continue
 
-        self.ota_base_station_id = (
-            0 if self.ota_base_station_id is None else self.ota_base_station_id
-        )
+        self.ota_base_station_id = 0 if self.ota_base_station_id is None else self.ota_base_station_id
 
         # Send a REQ (request) message through the OTA device
         # Next cycle we will hopefully have a CON (connected) response
-        self.ota.send_ota_message(
-            self.ota_base_station_id, "REQ"
-        )  # bUE sends a message back to the base station asking to
+        self.ota.send_ota_message(self.ota_base_station_id, "REQ")  # bUE sends a message back to the base station asking to
         # be connected as well
         print("Sending a REQ")
 
@@ -187,9 +177,7 @@ def ota_idle_ping(self):
         print("Getting GPS Data")
         lat, long = self.gps_handler()
         print("trying to send a ping")
-        self.ota.send_ota_message(
-            self.ota_base_station_id, f"PING,{lat},{long}"
-        )  # test ping for now
+        self.ota.send_ota_message(self.ota_base_station_id, f"PING,{lat},{long}")  # test ping for now
 
         # See if there are any new messages from the OTA device
         try:
@@ -215,9 +203,7 @@ def ota_idle_ping(self):
                 self.ota_timeout = TIMEOUT
                 got_pingr = True
 
-            elif (
-                "TEST" in message
-            ):  # Message should look like 1,34,TEST.<file>.<configuration>.<role>.<starttime>,-1,8
+            elif "TEST" in message:  # Message should look like 1,34,TEST.<file>.<configuration>.<role>.<starttime>,-1,8
                 input = parts[2]
 
                 self.test_handler(input)
@@ -236,13 +222,9 @@ def ota_idle_ping(self):
             self.ota_timeout -= 1
 
             if self.ota_timeout <= TIMEOUT / 2:
-                logger.info(
-                    f"We haven't heard from {self.ota_base_station_id} in a while...."
-                )
+                logger.info(f"We haven't heard from {self.ota_base_station_id} in a while....")
             if self.ota_timeout <= 0:
-                logger.info(
-                    f"We have not heard from {self.ota_base_station_id} in too long. Disconnecting..."
-                )
+                logger.info(f"We have not heard from {self.ota_base_station_id} in too long. Disconnecting...")
                 self.ota_connected = False
 
     """
@@ -284,9 +266,7 @@ def ota_idle_ping(self):
     #         logger.error(f"GPS error: {e}")    #     logger.debug("Could not find coordinates. Are they off?")
     #     return "", ""
 
-    def gps_handler(
-        self, max_attempts=50, min_fixes=3, hdop_threshold=2.0, max_runtime=10
-    ):
+    def gps_handler(self, max_attempts=50, min_fixes=3, hdop_threshold=2.0, max_runtime=10):
         start_time = time.time()
         try:
             session = gps.gps(mode=gps.WATCH_ENABLE | gps.WATCH_NEWSTYLE)
@@ -310,9 +290,7 @@ def gps_handler(
                             eph = getattr(report, "eph", None)
 
                             if lat is not None and lon is not None:
-                                logger.debug(
-                                    f"Got GPS fix: lat={lat}, lon={lon}, HDOP={eph}"
-                                )
+                                logger.debug(f"Got GPS fix: lat={lat}, lon={lon}, HDOP={eph}")
                                 all_fixes.append((lat, lon))
                             else:
                                 logger.debug("GPS fix missing lat/lon fields")
@@ -352,9 +330,7 @@ def test_handler(self, input):  # Input Format: TEST-<file>-<wait_time>-<paramet
                 current_time = int(time.time())
                 if start_time > current_time:
                     wait_duration = start_time - current_time
-                    logger.info(
-                        f"Waiting {wait_duration} seconds until start time {start_time}"
-                    )
+                    logger.info(f"Waiting {wait_duration} seconds until start time {start_time}")
 
                     # Wait in small increments to allow for cancellation
                     while int(time.time()) < start_time and not self.cancel_test:
@@ -382,9 +358,7 @@ def test_handler(self, input):  # Input Format: TEST-<file>-<wait_time>-<paramet
                     text=True,  # decode bytes to str
                 )
 
-                logger.info(
-                    f"Started test script: {file}.py with parameters {parameters}"
-                )
+                logger.info(f"Started test script: {file}.py with parameters {parameters}")
 
                 def reader_thread(pipe, output_queue):
                     for line in iter(pipe.readline, ""):
@@ -413,9 +387,7 @@ def reader_thread(pipe, output_queue):
 
                 # We also collect all the terminal outputs from the script so we can send them back to the base station
                 """
-                while (
-                    process.poll() is None
-                ):  # poll() returns None if process hasn't terminated
+                while process.poll() is None:  # poll() returns None if process hasn't terminated
                     # Get all normal terminal outputs
                     try:
                         stdout_line = stdout_queue.get_nowait()
@@ -514,22 +486,14 @@ def ota_send_upd(self):
         with self.test_output_lock:
             if self.test_output_buffer:
                 for line in self.test_output_buffer:
-                    self.ota.send_ota_message(
-                        self.ota_base_station_id, f"UPD:,{lat},{long},{line}"
-                    )
-                    logger.info(
-                        f"Sent UPD to {self.ota_base_station_id} with console output: {line}"
-                    )
+                    self.ota.send_ota_message(self.ota_base_station_id, f"UPD:,{lat},{long},{line}")
+                    logger.info(f"Sent UPD to {self.ota_base_station_id} with console output: {line}")
                     time.sleep(0.4)  # Sleep so UART does not get overwhelmed
                 self.test_output_buffer.clear()
 
             else:  # If there is no message send it blank
-                self.ota.send_ota_message(
-                    self.ota_base_station_id, f"UPD:,{lat},{long},"
-                )
-                logger.info(
-                    f"Sent UPD to {self.ota_base_station_id} with no console output"
-                )
+                self.ota.send_ota_message(self.ota_base_station_id, f"UPD:,{lat},{long},")
+                logger.info(f"Sent UPD to {self.ota_base_station_id} with no console output")
 
     """
     # This function checks for incoming messages while in the system is the UTW_TEST state.
@@ -553,9 +517,7 @@ def check_for_cancel(self):
                 logger.info(f"Received a RESTART message")
                 self.restart_system()
             else:
-                logger.error(
-                    f"Received unexpected message while in UTW_TEST state: {message}"
-                )
+                logger.error(f"Received unexpected message while in UTW_TEST state: {message}")
 
     """
     Restarts the service entirely
@@ -581,9 +543,7 @@ def restart_system(self):
 
     def state_change_logger(self):
         if self.cur_st != self.prv_st:
-            logger.info(
-                f"state_change_logger: State changed from {self.prv_st.name} to {self.cur_st.name}"
-            )
+            logger.info(f"state_change_logger: State changed from {self.prv_st.name} to {self.cur_st.name}")
             self.prv_st = self.cur_st
 
     def bue_tick(self, loop_dur=0.01):

bue_main.py

@@ -91,15 +91,11 @@ def send_test_to_bue(self, bue_id, test_script):
         """Simulate sending a test to a bUE (like PREPR message)"""
         if bue_id in self.connected_bues and bue_id not in self.testing_bues:
             self.testing_bues.append(bue_id)
-            self.stdout_history.append(
-                f"[test_manager.py] PREPR sent to bUE {bue_id} for {test_script}"
-            )
+            self.stdout_history.append(f"[test_manager.py] PREPR sent to bUE {bue_id} for {test_script}")
 
             # Simulate test completion after random time (5-15 seconds)
             completion_delay = random.uniform(5, 15)
-            completion_timer = threading.Timer(
-                completion_delay, self._complete_test, args=[bue_id]
-            )
+            completion_timer = threading.Timer(completion_delay, self._complete_test, args=[bue_id])
             completion_timer.daemon = True
             completion_timer.start()
 
@@ -108,17 +104,13 @@ def _complete_test(self, bue_id):
         if bue_id in self.testing_bues:
             self.testing_bues.remove(bue_id)
             completion_type = random.choice(["DONE", "FAIL"])
-            self.stdout_history.append(
-                f"[test_manager.py] Test {completion_type} on bUE {bue_id}"
-            )
+            self.stdout_history.append(f"[test_manager.py] Test {completion_type} on bUE {bue_id}")
 
     def cancel_test_on_bue(self, bue_id):
         """Simulate canceling a test on a bUE (like CANCD message)"""
         if bue_id in self.testing_bues:
             self.testing_bues.remove(bue_id)
-            self.stdout_history.append(
-                f"[test_manager.py] CANCD sent to bUE {bue_id} - test cancelled"
-            )
+            self.stdout_history.append(f"[test_manager.py] CANCD sent to bUE {bue_id} - test cancelled")
 
     def mock_updates(self):
         """Simulate dynamic updates to the base station data"""
@@ -186,9 +178,7 @@ def get_distance(self, bue_1, bue_2):
 
             # Convert to approximate meters (rough calculation)
             lat_meters = lat_diff * 111000  # 1 degree lat ≈ 111km
-            lon_meters = (
-                lon_diff * 111000 * 0.8
-            )  # Adjust for longitude at this latitude
+            lon_meters = lon_diff * 111000 * 0.8  # Adjust for longitude at this latitude
 
             distance = (lat_meters**2 + lon_meters**2) ** 0.5
             return distance