Use structured logging

lib/pysquared/Big_Data.py

@@ -1,14 +1,15 @@
 import gc
 
-from lib.pysquared.debugcolor import co
+from lib.pysquared.logger import Logger
 
 
 class Face:
-    def __init__(self, Add, Pos, debug_state, tca):
+    def __init__(self, Add, Pos, debug_state, tca, logger: Logger):
         self.tca = tca
         self.address = Add
         self.position = Pos
         self.debug = debug_state
+        self.logger = logger
 
         # Use tuple instead of list for immutable data
         self.senlist = ()
@@ -30,10 +31,6 @@ def __init__(self, Add, Pos, debug_state, tca):
         self.veml = None
         self.drv = None
 
-    def debug_print(self, statement):
-        if self.debug:
-            print(co("[FACE]" + statement, "teal", "bold"))
-
     def Sensorinit(self, senlist, address):
         gc.collect()  # Force garbage collection before initializing sensors
 
@@ -44,7 +41,7 @@ def Sensorinit(self, senlist, address):
                 self.mcp = adafruit_mcp9808.MCP9808(self.tca[address], address=27)
                 self.sensors["MCP"] = True
             except Exception as e:
-                self.debug_print("[ERROR][Temperature Sensor]" + str(e))
+                self.logger.error("Error Initializing Temperature Sensor", err=e)
 
         if "VEML" in senlist:
             try:
@@ -53,7 +50,7 @@ def Sensorinit(self, senlist, address):
                 self.veml = adafruit_veml7700.VEML7700(self.tca[address])
                 self.sensors["VEML"] = True
             except Exception as e:
-                self.debug_print("[ERROR][Light Sensor]" + str(e))
+                self.logger.error("Error Initializing Light Sensor", err=e)
 
         if "DRV" in senlist:
             try:
@@ -62,21 +59,22 @@ def Sensorinit(self, senlist, address):
                 self.drv = adafruit_drv2605.DRV2605(self.tca[address])
                 self.sensors["DRV"] = True
             except Exception as e:
-                self.debug_print("[ERROR][Motor Driver]" + str(e))
+                self.logger.error("Error Initializing Motor Driver", err=e)
 
         gc.collect()  # Clean up after initialization
 
 
 class AllFaces:
-    def __init__(self, debug, tca):
+    def __init__(self, debug, tca, logger: Logger):
         self.tca = tca
         self.debug = debug
         self.faces = []
+        self.logger = logger
 
         # Create faces using a loop instead of individual variables
         positions = [("y+", 0), ("y-", 1), ("x+", 2), ("x-", 3), ("z-", 4)]
         for pos, addr in positions:
-            face = Face(addr, pos, debug, tca)
+            face = Face(addr, pos, debug, tca, self.logger)
             face.Sensorinit(face.senlist, face.address)
             self.faces.append(face)
             gc.collect()  # Clean up after each face initialization

lib/pysquared/Field.py

@@ -4,42 +4,33 @@
 Authors: Nicole Maggard, Michael Pham, and Rachel Sarmiento
 """
 
-import traceback
-
-from lib.pysquared.debugcolor import co
+from lib.pysquared.logger import Logger
 
 
 class Field:
-    def debug_print(self, statement):
-        if self.debug:
-            print(co("[Field]" + statement, "pink", "bold"))
-
-    def __init__(self, cubesat, debug):
+    def __init__(self, cubesat, debug, logger: Logger):
         self.debug = debug
         self.cubesat = cubesat
+        self.logger = logger
 
     def Beacon(self, msg):
         try:
             if self.cubesat.is_licensed:
-                self.debug_print("I am beaconing: " + str(msg))
-                print(
-                    "Message Success: "
-                    + str(self.cubesat.radio1.send(bytes(msg, "UTF-8")))
+                self.logger.info(
+                    "I am beaconing",
+                    beacon=str(msg),
+                    success=str(self.cubesat.radio1.send(bytes(msg, "UTF-8"))),
                 )
             else:
-                self.debug_print(
-                    "Please toggle licensed variable in code once you obtain an amateur radio license"
+                self.logger.debug(
+                    "Please toggle licensed variable in code once you obtain an amateur radio license",
                 )
         except Exception as e:
-            self.debug_print(
-                "Tried Beaconing but encountered error: ".join(
-                    traceback.format_exception(e)
-                )
-            )
+            self.logger.error("There was an error while Beaconing", err=e)
 
     def troubleshooting(self):
         # this is for troubleshooting comms
         pass
 
     def __del__(self):
-        self.debug_print("Object Destroyed!")
+        self.logger.debug("Object Destroyed!")

lib/pysquared/battery_helper.py

@@ -1,5 +1,7 @@
 import time
 
+from lib.pysquared.logger import Logger
+
 # Written with Claude 3.5
 # Author: Michael Pham
 # Date: 2024-11-05
@@ -21,7 +23,7 @@ class BatteryHelper:
     CMD_RESET_MCU = "11"  # Reset microcontroller
     CMD_ERROR = "208"
 
-    def __init__(self, pysquared):
+    def __init__(self, pysquared, logger: Logger):
         """
         Initialize UART helper with existing Pysquared object
 
@@ -31,6 +33,7 @@ def __init__(self, pysquared):
         self.uart = pysquared.uart
         self.last_command_time = 0
         self.debug_mode = True
+        self.logger = logger
 
     def _flush_input(self):
         """Flush the input buffer"""
@@ -52,7 +55,7 @@ def _wait_for_ack(self):
             if self.uart.in_waiting:
                 byte = self.uart.read(1)
                 if self.debug_mode:
-                    print(f"ACK byte received: {byte}")
+                    self.logger.info("ACK received", ack=byte)
                 if byte == b"A":
                     return True
             time.sleep(0.001)
@@ -80,7 +83,7 @@ def _read_message(self, timeout_ms=150):
         try:
             text = response.decode("utf-8")
             if self.debug_mode:
-                print(f"Buffer: {text}")
+                self.logger.info("Read message", buffer=text)
 
             # Check for complete message
             if "AA<" in text and ">" in text:
@@ -89,8 +92,7 @@ def _read_message(self, timeout_ms=150):
                 if start_idx < end_idx:
                     return text[start_idx + 1 : end_idx]
         except Exception as e:
-            print(f"Error decoding message: {e}")
-            pass
+            self.logger.error("Error decoding message", err=e)
 
         return ""
 
@@ -109,7 +111,7 @@ def _send_command(self, cmd):
             return self._read_message()
 
         except Exception as e:
-            print(f"UART error: {e}")
+            self.logger.error("UART error", err=e)
             return ""
 
     def _is_valid_message(self, msg):
@@ -155,10 +157,10 @@ def get_power_metrics(self):
 
             except Exception as e:
                 if self.debug_mode:
-                    print(f"Error parsing metrics: {e}")
+                    self.logger.error("Error parsing metrics", err=e)
 
         if self.debug_mode:
-            print("Failed to get valid power metrics")
+            self.logger.warning("Failed to get valid power metrics")
         return (0.0, 0.0, 0.0, 0.0, False, 0.0)
 
     def get_error_metrics(self):
@@ -255,8 +257,6 @@ def get_battery_percentage(self, pack_voltage, is_charging=False):
     def debug_timing(self):
         """Measure and print timing of each step"""
 
-        print("\nTiming analysis:")
-
         # Measure command send time
         start = time.monotonic()
         self.uart.write(bytes(self.CMD_GET_POWER.encode()))
@@ -275,14 +275,17 @@ def debug_timing(self):
                 values = [float(x) for x in parts[:4]]
                 values.append(bool(int(parts[4])))
             except Exception as e:
-                print(f"Parse error: {e}")
+                self.logger.error("Parse error", err=e)
         parse_time = (time.monotonic() - parse_start) * 1000
 
         # Total time
         total_time = (time.monotonic() - start) * 1000
 
-        print(f"Send time: {send_time:.2f}ms")
-        print(f"Read time: {read_time:.2f}ms")
-        print(f"Parse time: {parse_time:.2f}ms")
-        print(f"Total time: {total_time:.2f}ms")
-        print(f"Response: {response}")
+        self.logger.info(
+            "timing analysis",
+            send_time=f"{send_time:.2f}ms",
+            read_time=f"{read_time:.2f}ms",
+            parse_time=f"{parse_time:.2f}ms",
+            total_time=f"{total_time:.2f}ms",
+            response=response,
+        )