Uart

.gitignore

@@ -16,4 +16,4 @@ bundles
 dist
 **/*.egg-info
 .vscode
-*notes*
+*notes*

adafruit_bno080/__init__.py

@@ -31,14 +31,13 @@
 from collections import namedtuple
 import time
 from micropython import const
-import digitalio
 
 # TODO: Remove on release
 from .debug import channels, reports
 
 # TODO: shorten names
 # Channel 0: the SHTP command channel
-_BNO_CHANNEL_SHTP_COMMAND = const(0)
+BNO_CHANNEL_SHTP_COMMAND = const(0)
 BNO_CHANNEL_EXE = const(1)
 _BNO_CHANNEL_CONTROL = const(2)
 _BNO_CHANNEL_INPUT_SENSOR_REPORTS = const(3)
@@ -134,7 +133,8 @@
 
 class PacketError(Exception):
     """Raised when the packet couldnt be parsed"""
-    pass
+
+    pass  # pylint:disable=unnecessary-pass
 
 
 def _elapsed(start_time):
@@ -145,11 +145,11 @@ def elapsed_time(func):
     """Print the runtime of the decorated function"""
 
     def wrapper_timer(*args, **kwargs):
-        start_time = time.monotonic_ns()  # 1
+        start_time = time.monotonic()  # 1
         value = func(*args, **kwargs)
-        end_time = time.monotonic_ns()  # 2
+        end_time = time.monotonic()  # 2
         run_time = end_time - start_time  # 3
-        print("Finished", func.__name__, "in", (run_time / 1000000.0), "ms")
+        print("Finished", func.__name__, "in", (run_time * 1000.0), "ms")
         return value
 
     return wrapper_timer
@@ -388,7 +388,9 @@ def geomagnetic_quaternion(self):
         try:
             return self._readings[BNO_REPORT_GEOMAGNETIC_ROTATION_VECTOR]
         except KeyError:
-            raise RuntimeError("No geomag quaternion report found, is it enabled?") from None
+            raise RuntimeError(
+                "No geomag quaternion report found, is it enabled?"
+            ) from None
 
     @property
     def steps(self):
@@ -450,15 +452,15 @@ def shake(self):
     def _process_available_packets(self):
         processed_count = 0
         while self._data_ready:
-            print("reading a packet")
+            # print("reading a packet")
             try:
                 new_packet = self._read_packet()
             except PacketError:
                 continue
             self._handle_packet(new_packet)
             processed_count += 1
             self._dbg("")
-            #print("Processed", processed_count, "packets")
+            # print("Processed", processed_count, "packets")
             self._dbg("")
             # we'll probably need an exit here for fast sensor rates
         self._dbg("")
@@ -480,8 +482,12 @@ def _wait_for_packet_type(self, channel_number, report_id=None, timeout=5.0):
                         return new_packet
                 else:
                     return new_packet
-            self._dbg("passing packet to handler for de-slicing")
-            self._handle_packet(new_packet)
+            if new_packet.channel_number not in (
+                BNO_CHANNEL_EXE,
+                BNO_CHANNEL_SHTP_COMMAND,
+            ):
+                self._dbg("passing packet to handler for de-slicing")
+                self._handle_packet(new_packet)
 
         raise RuntimeError("Timed out waiting for a packet on channel", channel_number)
 
@@ -503,11 +509,14 @@ def _update_sequence_number(self, new_packet):
         self._sequence_number[channel] = seq
 
     def _handle_packet(self, packet):
-
         # split out reports first
-        _separate_batch(packet, self._packet_slices)
-        while len(self._packet_slices) > 0:
-            self._process_report(*self._packet_slices.pop())
+        try:
+            _separate_batch(packet, self._packet_slices)
+            while len(self._packet_slices) > 0:
+                self._process_report(*self._packet_slices.pop())
+        except Exception as error:
+            print(packet)
+            raise error
 
     def _handle_control_report(self, report_id, report_bytes):
         if report_id == _SHTP_REPORT_PRODUCT_ID_RESPONSE:
@@ -543,8 +552,11 @@ def _process_report(self, report_id, report_bytes):
             if report_id == BNO_REPORT_SHAKE_DETECTOR:
                 shake_detected = _parse_shake_report(report_bytes)
                 # shake not previously detected - auto cleared by 'shake' property
-                if not self._readings[BNO_REPORT_SHAKE_DETECTOR]:
-                    self._readings[BNO_REPORT_SHAKE_DETECTOR] = shake_detected
+                try:
+                    if not self._readings[BNO_REPORT_SHAKE_DETECTOR]:
+                        self._readings[BNO_REPORT_SHAKE_DETECTOR] = shake_detected
+                except KeyError:
+                    pass
                 return
 
             sensor_data = _parse_sensor_report_data(report_bytes)
@@ -567,11 +579,13 @@ def _get_feature_enable_report(
         pack_into("<I", set_feature_report, 5, report_interval)
         return set_feature_report
 
+    # TODO: add docs for available features
     def enable_feature(self, feature_id):
+        """Used to enable a given feature of the BNO080"""
         self._dbg("\n********** Enabling feature id:", feature_id, "**********")
 
         set_feature_report = self._get_feature_enable_report(feature_id)
-        print("Enabling", feature_id)
+        # print("Enabling", feature_id)
         self._send_packet(_BNO_CHANNEL_CONTROL, set_feature_report)
         while True:
             packet = self._wait_for_packet_type(
@@ -585,11 +599,9 @@ def enable_feature(self, feature_id):
                     self._readings[feature_id] = (0.0, 0.0, 0.0, 0.0)
                 else:
                     self._readings[feature_id] = (0.0, 0.0, 0.0)
-                print("Enabled", feature_id)
+                # print("Enabled", feature_id)
                 break
-            else:
-                raise RuntimeError("Was not able to enable feature", feature_id)
-
+            raise RuntimeError("Was not able to enable feature", feature_id)
 
     def _check_id(self):
         self._dbg("\n********** READ ID **********")
@@ -641,7 +653,6 @@ def _get_data(self, index, fmt_string):
         data_index = index + 4
         return unpack_from(fmt_string, self._data_buffer, offset=data_index)[0]
 
-
     # pylint:disable=no-self-use
     @property
     def _data_ready(self):
@@ -651,7 +662,9 @@ def hard_reset(self):
         """Hardware reset the sensor to an initial unconfigured state"""
         if not self._reset:
             return
-        #print("Hard resetting...")
+        # print("Hard resetting...")
+        import digitalio  # pylint:disable=import-outside-toplevel
+
         self._reset.direction = digitalio.Direction.OUTPUT
         self._reset.value = True
         time.sleep(0.01)
@@ -665,25 +678,18 @@ def soft_reset(self):
         print("Soft resetting...", end="")
         data = bytearray(1)
         data[0] = 1
-        seq = self._send_packet(BNO_CHANNEL_EXE, data)
+        _seq = self._send_packet(BNO_CHANNEL_EXE, data)
         time.sleep(0.5)
-
-        for i in range(3):
-            while True:  # retry reading packets until ready!
-                try:
-                    packet = self._read_packet()
-                    break
-                except PacketError:
-                    time.sleep(0.1)
-
-            #print(packet)
-            if i == 0 and packet.channel_number != _BNO_CHANNEL_SHTP_COMMAND:
-                raise RuntimeError("Expected an SHTP announcement")
-            if i == 1 and packet.channel_number != BNO_CHANNEL_EXE:
-                raise RuntimeError("Expected a reset reply")
-            if i == 2 and packet.channel_number != _BNO_CHANNEL_CONTROL:
-                raise RuntimeError("Expected a control announcement")
-        print("OK!");
+        _seq = self._send_packet(BNO_CHANNEL_EXE, data)
+        time.sleep(0.5)
+
+        for _i in range(3):
+            try:
+                _packet = self._read_packet()
+            except PacketError:
+                time.sleep(0.5)
+
+        print("OK!")
         # all is good!
 
     def _send_packet(self, channel, data):

adafruit_bno080/i2c.py

@@ -6,12 +6,9 @@
     Subclass of `adafruit_bno080.BNO080` to use I2C
 
 """
-import time
 from struct import pack_into
 import adafruit_bus_device.i2c_device as i2c_device
-from . import BNO080, BNO_CHANNEL_EXE, DATA_BUFFER_SIZE, const, Packet, PacketError
-
-# should be removeable; I _think_ something else should be able to prep the buffers?
+from . import BNO080, DATA_BUFFER_SIZE, const, Packet, PacketError
 
 _BNO080_DEFAULT_ADDRESS = const(0x4A)
 
@@ -23,7 +20,9 @@ class BNO080_I2C(BNO080):
 
     """
 
-    def __init__(self, i2c_bus, reset=None, address=_BNO080_DEFAULT_ADDRESS, debug=False):
+    def __init__(
+        self, i2c_bus, reset=None, address=_BNO080_DEFAULT_ADDRESS, debug=False
+    ):
         self.bus_device_obj = i2c_device.I2CDevice(i2c_bus, address)
         super().__init__(reset, debug)
 
@@ -33,10 +32,7 @@ def _send_packet(self, channel, data):
 
         pack_into("<H", self._data_buffer, 0, write_length)
         self._data_buffer[2] = channel
-
         self._data_buffer[3] = self._sequence_number[channel]
-
-        # this is dumb but it's what we have for now
         for idx, send_byte in enumerate(data):
             self._data_buffer[4 + idx] = send_byte
 
@@ -58,12 +54,10 @@ def _read_header(self):
         return packet_header
 
     def _read_packet(self):
-        # TODO: MAGIC NUMBER?
-        # TODO: this can be `_read_header` or know it's done by `_data_ready`
         with self.bus_device_obj as i2c:
             i2c.readinto(self._data_buffer, end=4)  # this is expecting a header?
         self._dbg("")
-        #print("SHTP READ packet header: ", [hex(x) for x in self._data_buffer[0:4]])
+        # print("SHTP READ packet header: ", [hex(x) for x in self._data_buffer[0:4]])
 
         header = Packet.header_from_buffer(self._data_buffer)
         packet_byte_count = header.packet_byte_count
@@ -85,7 +79,6 @@ def _read_packet(self):
 
         self._read(packet_byte_count)
 
-        # TODO: Allocation
         new_packet = Packet(self._data_buffer)
         if self._debug:
             print(new_packet)
@@ -114,7 +107,6 @@ def _data_ready(self):
 
         if header.channel_number > 5:
             self._dbg("channel number out of range:", header.channel_number)
-        #  data_length, packet_byte_count)
         if header.packet_byte_count == 0x7FFF:
             print("Byte count is 0x7FFF/0xFFFF; Error?")
             if header.sequence_number == 0xFF: