Running in debug mode for longer periods

[bastiaanterhorst]

Hi,
My application relies on reading raw values from the sensor, so I am running in debug mode. This works very well, except that after a few days of use, something seems to break the data that the sensor is sending through. I get garbage that I am unable to decode (as UTF-8, following the sample Python code in this repository).

I have tried restarting the serial connection and completely unloading and reloading the USB serial driver, but neither reliably gets the sensor in a working state again.

The only thing that reliably works is going through a full power cycle, which is inconvenient for my application.

I was wondering:
a) is this to be expected? will debug mode only work for short periods of time?
b) do you have any tips to make the debug connection work for longer?
c) is there a way to power cycle or reset the sensor in some way, via the serial connection through the electronics circuit?

Below follows the complete class I wrote to use the sensor. This relies on other components so will not run in isolation, but it does illustrate my use of the sensor, in case it helps to answer my query. Thanks!

"""
This class handles communication with an MW0582 microwave doppler radar sensor.
The sensor is used in debug mode, which gives access to its raw readings. These readings
are used to trigger movement detection in close (action) and distant (motion) proximity.
Signals are triggered for both of these events.
"""
from lib.singleton import Singleton
import serial, threading, re, time
import serial.tools.list_ports as prtlst
import numpy as np
import os
from blinker import signal

class MotionSensor(metaclass=Singleton):
  """ This class implements the motion sensor."""

  # Signal thresholds for motion and action detection
  SIG_MOTION = 150
  SIG_ACTION = 700

  # Number of seconds to ignore data after a trigger, to debounce events
  SIG_DEBOUNCE = 2

  def __init__(self):
    """ 
    Start monitoring the sensor in a thread
    """

    # start reading data from the device in a separate thread
    self.ser_thread = threading.Thread(target=self.read_thread, name="Motion Sensor", daemon=True)
    self.ser_thread.start()

    # These are used to store the time and type of the last signal
    self.s_time = 0
    self.s_type = self.SIG_MOTION

  def __exit__(self, exc_type, exc_value, exc_traceback):
    """ Close the connection to sensor on exit."""
    # Turn off debug mode before exiting
    self.ser.write(b'AT+DEBUG=0000')
    self.ser.close()

  def find_device(self):
    """
    Finds the first USB tty device, so we can connect to it.
    """
    ports = prtlst.comports()
    for port in ports:
      if 'USB' in port[1]: #check 'USB' string in device description
        return port[0]

  def read_thread(self):
    """
    Reads from the serial device and triggers close or distant motion events.
    Events are debounced using a timer. When a motion event is detected, it can be
    superseded by a subsequent action event. Actions events cannot be supersedeld, until
    the debouce timer has lapsed. Current values are also stored so they can be fetched
    raw for use outside of the standard events.
    """
    # Before anything, find the device
    device = self.find_device()

    # First set up the serial connection to the sensor
    self.ser = serial.Serial(device, 512000, timeout=0)
    self.ser.flush()

    # Put the device in debug mode
    self.ser.write(b'AT+DEBUG=0002')

    # let the device settle and the buffer fill
    time.sleep(1.0)

    errors = 0
    error_budget = 20

    while True:
      try:
        line = self.ser.read(1024)
        try:
          data = [int(x, 16) for x in re.findall(' (?!f)\w{4}', line.decode('utf-8'))]
          new_data = abs(data[-1]-1000)

          # store the measurement so we can read it in real time
          self.last_measurement = new_data

          # see if we are in a debounce
          db = not (time.time() > self.s_time + self.SIG_DEBOUNCE)

          # if we are in a debounce for a motion event, and we've now found an action event: trigger it
          if db == True and self.s_type == self.SIG_MOTION and new_data > self.SIG_ACTION:
            # action overrides motion event
            self.send_signal(self.SIG_ACTION) # for some reason this is not being called
          # otherwise we only care about stuff happening outside the debounce period
          elif db == False:
            if new_data > self.SIG_ACTION:
              self.send_signal(self.SIG_ACTION)
            elif new_data > self.SIG_MOTION:
              self.send_signal(self.SIG_MOTION)
          
          # wait a little while so we don't saturate the CPU
          time.sleep(0.1)
          
          # reset our error counter if we reach the end here successfully
          errors = 0
        except Exception as e:
          # on every exception we count the error. If we reach our error budget we break out of the loop to force a reset.
          errors = errors + 1
          print("Motion sensor exception", e, errors, error_budget)
          if errors > error_budget:
            break
      except KeyboardInterrupt:
        pass

    self.re_prime()
    self.read_thread()

  def re_prime(self):
    """ 
    Closes the current serial connection and resets the USB device, to prepare for a fresh connection.
    This is used to fix flakiness in the serial connection.
    """
    print("Re-priming serial connection")

    # exit out of debug mode, close the connection, wait a little while, and then try again
    self.ser.write(b'AT+DEBUG=0000')
    self.ser.close()
    time.sleep(0.1)

    # Unbind the USB device, unload the device driver, unload the USB Serial driver, wait a little, reload everything using modprobe
    os.system("rmmod ch341")
    os.system("rmmod usbserial ")
    time.sleep(0.1)
    os.system("modprobe ch341")

  def read(self):
    return self.last_measurement

  def send_signal(self, s_type):
    self.s_time = time.time()
    self.s_type = s_type
    if s_type == self.SIG_ACTION:
      sig = signal('motion_sense_action')
    else:
      sig = signal('motion_sense_motion')
    sig.send()