python lib: print() -> logging (#1215)

* python lib: print() -> logging

* fstrings are fine

.pre-commit-config.yaml

@@ -32,5 +32,5 @@ repos:
         - -rn
         - -sn
         - -j0
-        - --disable=C,R,W0613,W0511,W0212,W0201,W0311,W0106,W0603,W0621,W0703,W1514,E1136
+        - --disable=C,R,W0613,W0511,W0212,W0201,W0311,W0106,W0603,W0621,W0703,W1203,W1514,E1136
         - --generated-members="usb1.*"

python/__init__.py

@@ -27,8 +27,6 @@ logging.basicConfig(level=LOGLEVEL, format='%(message)s')
 
 BASEDIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "../")
 
-DEBUG = os.getenv("PANDADEBUG") is not None
-
 USBPACKET_MAX_SIZE = 0x40
 CANPACKET_HEAD_SIZE = 0x6
 DLC_TO_LEN = [0, 1, 2, 3, 4, 5, 6, 7, 8, 12, 16, 20, 24, 32, 48, 64]
@@ -264,7 +262,7 @@ class Panda:
     assert self._handle is not None
     self._mcu_type = self.get_mcu_type()
     self.health_version, self.can_version, self.can_health_version = self.get_packets_versions()
-    print("connected")
+    logging.debug("connected")
 
     # disable openpilot's heartbeat checks
     if self._disable_checks:
@@ -349,7 +347,7 @@ class Panda:
         success = True
         break
       except Exception:
-        print("reconnecting is taking %d seconds..." % (i + 1))
+        logging.debug("reconnecting is taking %d seconds...", i + 1)
         try:
           dfu = PandaDFU(PandaDFU.st_serial_to_dfu_serial(self._serial, self._mcu_type))
           dfu.recover()
@@ -374,22 +372,22 @@ class Panda:
     assert last_sector < 7, "Binary too large! Risk of overwriting provisioning chunk."
 
     # unlock flash
-    print("flash: unlocking")
+    logging.warning("flash: unlocking")
     handle.controlWrite(Panda.REQUEST_IN, 0xb1, 0, 0, b'')
 
     # erase sectors
-    print(f"flash: erasing sectors 1 - {last_sector}")
+    logging.warning(f"flash: erasing sectors 1 - {last_sector}")
     for i in range(1, last_sector + 1):
       handle.controlWrite(Panda.REQUEST_IN, 0xb2, i, 0, b'')
 
     # flash over EP2
     STEP = 0x10
-    print("flash: flashing")
+    logging.warning("flash: flashing")
     for i in range(0, len(code), STEP):
       handle.bulkWrite(2, code[i:i + STEP])
 
     # reset
-    print("flash: resetting")
+    logging.warning("flash: resetting")
     try:
       handle.controlWrite(Panda.REQUEST_IN, 0xd8, 0, 0, b'')
     except Exception:
@@ -399,7 +397,7 @@ class Panda:
     if not fn:
       fn = DEFAULT_H7_FW_FN if self._mcu_type == MCU_TYPE_H7 else DEFAULT_FW_FN
     assert os.path.isfile(fn)
-    print("flash: main version is " + self.get_version())
+    logging.debug("flash: main version is %s", self.get_version())
     if not self.bootstub:
       self.reset(enter_bootstub=True)
     assert(self.bootstub)
@@ -409,7 +407,7 @@ class Panda:
         code = f.read()
 
     # get version
-    print("flash: bootstub version is " + self.get_version())
+    logging.debug("flash: bootstub version is %s", self.get_version())
 
     # do flash
     Panda.flash_static(self._handle, code, mcu_type=self._mcu_type)
@@ -440,7 +438,7 @@ class Panda:
   def wait_for_dfu(dfu_serial: str, timeout: Optional[int] = None) -> bool:
     t_start = time.monotonic()
     while dfu_serial not in PandaDFU.list():
-      print("waiting for DFU...")
+      logging.debug("waiting for DFU...")
       time.sleep(0.1)
       if timeout is not None and (time.monotonic() - t_start) > timeout:
         return False
@@ -544,8 +542,8 @@ class Panda:
   def enter_bootloader(self):
     try:
       self._handle.controlWrite(Panda.REQUEST_OUT, 0xd1, 0, 0, b'')
-    except Exception as e:
-      print(e)
+    except Exception:
+      logging.exception("exception while entering bootloader")
 
   def get_version(self):
     return self._handle.controlRead(Panda.REQUEST_IN, 0xd6, 0, 0, 0x40).decode('utf8')
@@ -699,10 +697,10 @@ class Panda:
             tx = tx[bs:]
             if len(tx) == 0:
               break
-            print("CAN: PARTIAL SEND MANY, RETRYING")
+            logging.error("CAN: PARTIAL SEND MANY, RETRYING")
         break
       except (usb1.USBErrorIO, usb1.USBErrorOverflow):
-        print("CAN: BAD SEND MANY, RETRYING")
+        logging.error("CAN: BAD SEND MANY, RETRYING")
 
   def can_send(self, addr, dat, bus, timeout=CAN_SEND_TIMEOUT_MS):
     self.can_send_many([[addr, None, dat, bus]], timeout=timeout)
@@ -715,7 +713,7 @@ class Panda:
         dat = self._handle.bulkRead(1, 16384) # Max receive batch size + 2 extra reserve frames
         break
       except (usb1.USBErrorIO, usb1.USBErrorOverflow):
-        print("CAN: BAD RECV, RETRYING")
+        logging.error("CAN: BAD RECV, RETRYING")
         time.sleep(0.1)
     msgs, self.can_rx_overflow_buffer = unpack_can_buffer(self.can_rx_overflow_buffer + dat)
     return msgs
@@ -773,19 +771,15 @@ class Panda:
   # pulse low for wakeup
   def kline_wakeup(self, k=True, l=True):
     assert k or l, "must specify k-line, l-line, or both"
-    if DEBUG:
-      print("kline wakeup...")
+    logging.debug("kline wakeup...")
     self._handle.controlWrite(Panda.REQUEST_OUT, 0xf0, 2 if k and l else int(l), 0, b'')
-    if DEBUG:
-      print("kline wakeup done")
+    logging.debug("kline wakeup done")
 
   def kline_5baud(self, addr, k=True, l=True):
     assert k or l, "must specify k-line, l-line, or both"
-    if DEBUG:
-      print("kline 5 baud...")
+    logging.debug("kline 5 baud...")
     self._handle.controlWrite(Panda.REQUEST_OUT, 0xf4, 2 if k and l else int(l), addr, b'')
-    if DEBUG:
-      print("kline 5 baud done")
+    logging.debug("kline 5 baud done")
 
   def kline_drain(self, bus=2):
     # drain buffer
@@ -794,8 +788,7 @@ class Panda:
       ret = self._handle.controlRead(Panda.REQUEST_IN, 0xe0, bus, 0, 0x40)
       if len(ret) == 0:
         break
-      elif DEBUG:
-        print(f"kline drain: 0x{ret.hex()}")
+      logging.debug(f"kline drain: 0x{ret.hex()}")
       bret += ret
     return bytes(bret)
 
@@ -803,8 +796,8 @@ class Panda:
     echo = bytearray()
     while len(echo) != cnt:
       ret = self._handle.controlRead(Panda.REQUEST_OUT, 0xe0, bus, 0, cnt - len(echo))
-      if DEBUG and len(ret) > 0:
-        print(f"kline recv: 0x{ret.hex()}")
+      if len(ret) > 0:
+        logging.debug(f"kline recv: 0x{ret.hex()}")
       echo += ret
     return bytes(echo)
 
@@ -814,14 +807,13 @@ class Panda:
       x += bytes([sum(x) % 0x100])
     for i in range(0, len(x), 0xf):
       ts = x[i:i + 0xf]
-      if DEBUG:
-        print(f"kline send: 0x{ts.hex()}")
+      logging.debug(f"kline send: 0x{ts.hex()}")
       self._handle.bulkWrite(2, bytes([bus]) + ts)
       echo = self.kline_ll_recv(len(ts), bus=bus)
       if echo != ts:
-        print(f"**** ECHO ERROR {i} ****")
-        print(f"0x{echo.hex()}")
-        print(f"0x{ts.hex()}")
+        logging.error(f"**** ECHO ERROR {i} ****")
+        logging.error(f"0x{echo.hex()}")
+        logging.error(f"0x{ts.hex()}")
     assert echo == ts
 
   def kline_recv(self, bus=2, header_len=4):