Refactor to modules/classes for unit testing (#5)

* initial - back to motor command bits barely recognized

* motor bits displayed correctly, checksum passing
failing on telem

* tidyup, move telem data

* telem bits rxed again, no reset

* Split modules off into package for unit testing

Author: hyp0dermik-code

pd.py

@@ -23,37 +23,25 @@
 import sigrokdecode as srd
 from functools import reduce
 from enum import Enum
+from dshot.protoDshot import DshotCmd, DshotTelem, BitDshot, DshotSettings, Bit_DshotTelem
+
 
-gcr_tables = {
-    "0b11001": 0x0,
-    "0b11011": 0x1,
-    "0b10010": 0x2,
-    "0b10011": 0x3,
-    "0b11101": 0x4,
-    "0b10101": 0x5,
-    "0b10110": 0x6,
-    "0b10111": 0x7,
-    "0b11010": 0x8,
-    "0b1001": 0x9,
-    "0b1010": 0xa,
-    "0b1011": 0xb,
-    "0b11110": 0xc,
-    "0b1101": 0xd,
-    "0b1110": 0xe,
-    "0b1111": 0xf
-}
 
 class SamplerateError(Exception):
     pass
 
 class State(Enum):
-    CMD = 1,
+    RESET = 0
+    CMD = 1
     TELEM = 2
 
+
 class State_Telem(Enum):
-    START = 1,
+    RESET = 0
+    START = 1
     RECV = 2
 
+
 class Decoder(srd.Decoder):
     api_version = 3
     id = 'dshot'
@@ -106,200 +94,100 @@ def __init__(self):
 
     def reset(self):
         self.state = State.CMD
-        self.samplerate = None
-
-        self.debug = False
+        self.state_telem = State_Telem.START
 
+        self.samplerate = None
         self.inreset = False
-        self.bidirectional = False
-        self.dshot_kbaud = 300e3
-        self.dshot_period = 3.33e-6
-        self.actual_period = None
-        self.halfbitwidth = None
         self.currbit_ss = None
         self.currbit_es = None
-        self.samples_after_motorcmd = None
-        self.samples_pp = None
 
-        self.telem_start = None
-        self.state_telem = State_Telem.START
-        self.telem_baudrate_midpoint = 0
-        self.edt_force = False
+        self.dshot_cfg = DshotSettings()
+
+        self.debug = False
 
     def start(self):
-        self.bidirectional = True if self.options['bidir'] == 'True' else False
-        self.edt_force = True if self.options['edt_force'] == 'True' else False
-        self.dshot_kbaud = int(self.options['dshot_rate'])*1000
-        self.dshot_period = 1/self.dshot_kbaud
-        self.samples_pp =  int(self.samplerate*self.dshot_period)
-        self.samples_after_motorcmd = self.samples_pp * 3
-        self.samples_after_telempkt = self.samples_pp * 3
+        self.dshot_cfg.bidirectional = True if self.options['bidir'] == 'True' else False
+        self.dshot_cfg.edt_force = True if self.options['edt_force'] == 'True' else False
+        self.dshot_cfg.dshot_kbaud = int(self.options['dshot_rate'])*1000
+        self.dshot_cfg.samplerate = self.samplerate
+        self.dshot_cfg.update()
 
         self.out_ann = self.register(srd.OUTPUT_ANN)
-        self.telem_baudrate_midpoint = int((self.samplerate / (self.dshot_kbaud*(5/4))) / 2.0)
-        if self.debug:
-            print("telem_midpoint",self.telem_baudrate_midpoint)
 
     def metadata(self, key, value):
         if key == srd.SRD_CONF_SAMPLERATE:
             self.samplerate = value
 
-    def handle_bits_dshot(self, results):
-        #ss, es, bit
-        bits = [result[2] for result in results]
-
-        if len(bits) == 16:
-            dshot_value = int(reduce(lambda a, b: (a << 1) | b, bits[:11]))
-            telem_request = bits[11]
-            received_crc = int(reduce(lambda a, b: (a << 1) | b, bits[12:]))
-        
-            value_tocrc = int(reduce(lambda a, b: (a << 1) | b, bits[:12]))
-
-            if self.bidirectional:
-                calculated_crc = int((~(value_tocrc ^ (value_tocrc >> 4) ^ (value_tocrc >> 8)))&0x0F)
-            else:
-                calculated_crc = int(((value_tocrc ^ (value_tocrc >> 4) ^ (value_tocrc >> 8)))&0x0F)
-
-            if received_crc == calculated_crc:
-                crc_ok = True 
-            else:
-                crc_ok = False
-
-            # TODO: Align this correctly
-            crc_startsample = results[12][0]
-            
-            # Split annotation based on value type
-            if dshot_value < 48:
-                # Command
-                self.put(results[0][0], crc_startsample, self.out_ann,
-                        [1, ['%04d' % dshot_value]])
-            else:
-                # Throttle
-                 self.put(results[0][0], crc_startsample, self.out_ann,
-                        [2, ['%04d' % dshot_value]])
-
-            self.put(crc_startsample, results[15][1], self.out_ann, [3, ['Calc CRC: '+('%04d' % calculated_crc)+' TXed CRC:'+('%04d' % received_crc)]])
-            if not crc_ok:
-                self.put(crc_startsample, results[15][1], self.out_ann,
-                     [4, ['CRC INVALID']])
-            return True
-        else:
-            return False
-
-    def handle_bit_dshot(self, ss, es, nb_ss):
-        period = nb_ss - ss
-        duty = es - ss
-        # Ideal duty for T0H: 33%, T1H: 66%.
-        bit_ = (duty / period) > 0.5
-
-        self.put(ss, nb_ss, self.out_ann,
-        [0, ['%d' % bit_]])
-        return [ss,nb_ss,bit_]
-
-    def handle_telem_bit(self,matched):
-        # None to raise exception if no match
-        result = None
-
-        # Low
-        if matched == (True, False):
-            # 0 value
-            result = 0
-
-        # High
-        elif matched == (False, True):
-            # 1 value
-            result = 1
-        return result
-
-    def process_telem_erpm(self,packet,start,end):
-        # Raw packet
-        self.put(start,
-                 end, self.out_ann,
-                 [6, ['%23s' % bin(packet)]])
-        # XOR with next?
-        packet &= 0x0FFFFF
-        packet = (packet^(packet>>1))
-        self.put(start,
-                 end, self.out_ann,
-                 [7, ['%23s' % bin(packet)]])
-        # Undo GCR
-        output = 0b0
-
-        nibbles = 4
-        bitmask = 0b11111 << ((nibbles-1)*5)
-
-        for n in range(nibbles):
-            gcr_n = bitmask & packet
-            ungcr = gcr_tables[bin(gcr_n >> (nibbles - (n + 1)) * 5)]
-            output = (output << 4) | ungcr
-            bitmask = (bitmask >> 5)
-
-        # Compare CRC
-        crc_received = output & 0xF
-        output = (output >> 4) & 0xFFF
-        crc_calc = ~((output ^ (output >> 4) ^ (output >> 8))) & 0x0F
-
-
-
-        self.put(end - ((self.telem_baudrate_midpoint * 2)*4),
-                 end, self.out_ann,
-                 [7, ['%23s' % ("RX CRC: "+hex(crc_received)+" Calc CRC: "+hex(crc_calc))]])
-        if crc_calc != crc_received:
-            self.put(end - ((self.telem_baudrate_midpoint * 2) * 4),
-                     end, self.out_ann,
-                     [8, ['%23s' % ("CRC ERROR!")]])
-        # The upper 12 bit contain the eperiod (1/erps) in the following bitwise encoding:
-        #
-        # e e e m m m m m m m m m
-        #
-        # The 9 bit value M needs to shifted left E times to get the period in micro seconds.
-        # This gives a range of 1 us to 65408 us. Which translates to a min e-frequency of 15.29 hz or for 14 pole motors 3.82 hz.
-        return
-    def process_telem_edt(self,packet,start,end):
-
-        return
-    def process_telem(self,packet,start,end):
-        if self.edt_force:
-            self.process_telem_edt(packet,start,end)
-        else:
-            self.process_telem_erpm(packet,start,end)
 
-        return
 
+    def display_dshot(self,dshot):
+        crc_startsample = dshot.results[12].ss
+
+        # Split annotation based on value type
+        if dshot.dshot_value < 48:
+            # Command
+            self.put(dshot.results[0].ss, crc_startsample, self.out_ann,
+                     [1, ['%04d' % dshot.dshot_value]])
+        else:
+            # Throttle
+            self.put(dshot.results[0].ss, crc_startsample, self.out_ann,
+                     [2, ['%04d' % dshot.dshot_value]])
+
+        self.put(crc_startsample, dshot.results[15].es, self.out_ann,
+                 [3, ['Calc CRC: ' + ('%04d' % dshot.crc_calc) + ' TXed CRC:' + ('%04d' % dshot.crc_recv)]])
+        if not dshot.crc_ok:
+            self.put(crc_startsample, dshot.results[15].es, self.out_ann,
+                     [4, ['CRC INVALID']])
 
 
+    def complete_DshotBit(self, *args):
+        bitseq = BitDshot(*args)
+
+        self.put(bitseq.ss, bitseq.es, self.out_ann,
+                 [0, ['%d' % bool(bitseq)]])
+        return [bitseq]
 
     def decode(self):
         if not self.samplerate:
             raise SamplerateError('Cannot decode without samplerate.')
-        
+
+        dshot_value = DshotCmd(self.dshot_cfg)
+        telem_value = DshotTelem(self.dshot_cfg)
         results = []
         telem = 0b0
         tlm_start = 0
+
+        #bitseq = BitDshot()
         while True:
 
             match self.state:
                 case State.CMD:
-                    if not self.bidirectional:
-                        pins = self.wait([{0: 'r'}, {0: 'f'}, {'skip': self.samples_after_motorcmd}])
+                    if not self.dshot_cfg.bidirectional:
+                        pins = self.wait([{0: 'r'}, {0: 'f'}, {'skip': self.dshot_cfg.samples_after_motorcmd}])
                     else:
-                        pins = self.wait([{0: 'f'}, {0: 'r'}, {'skip': self.samples_after_motorcmd}])
+                        pins = self.wait([{0: 'f'}, {0: 'r'}, {'skip': self.dshot_cfg.samples_after_motorcmd}])
                     #TODO: Increase skip to maximum time for effiency
                     #TODO: Mark any changes in this time as errors?  Option to reduce load?
 
                     if self.currbit_ss and self.currbit_es and self.matched[2]:
                         # Assume end of packet if have seen start and end of a potential bit but no further change within 3 periods
                         # TODO: Confirm wait period this works with spec
-                        results += [self.handle_bit_dshot(self.currbit_ss, self.currbit_es,
-                                                          (self.currbit_ss + self.samples_pp))]
+
+                        args = self.currbit_ss, self.currbit_es, (self.currbit_ss + self.dshot_cfg.samples_pp)
+                        results += self.complete_DshotBit(*args)
                         self.currbit_ss = None
                         self.currbit_es = None
-
+                        #print(results)
                         # Pass results to decoder
-                        result = self.handle_bits_dshot(results)
-                        if result and self.bidirectional:
+
+                        result = dshot_value.handle_bits_dshot(results)
+                        if result:
+                            self.display_dshot(dshot_value)
+                        if result and self.dshot_cfg.bidirectional:
                             self.state = State.TELEM
+
                         results = []
+                        #dshot_value = DshotCmd(self.dshot_cfg)
 
                     if self.matched[0] and not self.currbit_ss and not self.currbit_es:
                         # Start of bit
@@ -309,13 +197,17 @@ def decode(self):
                         self.currbit_es = self.samplenum
                     elif self.matched[0] and self.currbit_es and self.currbit_ss:
                         # Have complete bit, can handle bit now
-                        result = [self.handle_bit_dshot(self.currbit_ss, self.currbit_es, self.samplenum)]
-                        # print(result)
-                        results += result
+                        args = self.currbit_ss, self.currbit_es, self.samplenum
+                        results += self.complete_DshotBit(*args)
+
                         self.currbit_ss = self.samplenum
                         self.currbit_es = None
                 case State.TELEM:
                     match self.state_telem:
+                        case State_Telem.RESET:
+                            telem = 0b0
+                            self.state_telem = State_Telem.START
+
                         case State_Telem.START:
                             # First wait for falling edge (idle high)
                             pins = self.wait([{0: 'f'}])
@@ -327,35 +219,36 @@ def decode(self):
                         case State_Telem.RECV:
                             # First conditions skips half bit width and matches low
                             # Second condition skips half bit width and matches high
-                            pins = self.wait([{0: 'l', 'skip': self.telem_baudrate_midpoint},
-                                              {0: 'h', 'skip': self.telem_baudrate_midpoint}])
+                            pins = self.wait([{0: 'l', 'skip': self.dshot_cfg.telem_baudrate_midpoint},
+                                              {0: 'h', 'skip': self.dshot_cfg.telem_baudrate_midpoint}])
 
                             # Append next bit
-                            curr_bit = self.handle_telem_bit(self.matched)
-                            self.put(self.samplenum - self.telem_baudrate_midpoint,
-                                     self.samplenum + self.telem_baudrate_midpoint,
+                            args = (self.samplenum - self.dshot_cfg.telem_baudrate_midpoint), self.samplenum, (self.samplenum + self.dshot_cfg.telem_baudrate_midpoint)
+                            curr_bit = Bit_DshotTelem(*args,self.matched)
+                            self.put(curr_bit.ss,curr_bit.es,
                                      self.out_ann,
-                                     [5, ['%04d' % curr_bit]])
-                            telem = telem | curr_bit
+                                     [5, ['%04d' % curr_bit.bit_]])
+
+                            telem_value.add_bit(curr_bit)
+
 
                             # Skip half bitwidth to end of bit
-                            pins = self.wait([{'skip': self.telem_baudrate_midpoint}])
+                            pins = self.wait([{'skip': self.dshot_cfg.telem_baudrate_midpoint}])
 
-                            if telem.bit_length() >= 20-1:
-                                self.process_telem(telem,tlm_start,self.samplenum)
+                            if telem_value.bits.bit_length() >= 20:
+                                telem_value.process_telem()
 
                                 # Reset
-                                telem = 0b0
-                                self.state_telem = State_Telem.START
+                                self.state_telem = State_Telem.RESET
+                                # Except Dshot packet next
                                 self.state = State.CMD
-                            else:
-                                # Shift for next bit
-                                telem = telem << 1
+
                             # If not mark as error
 
                             # Then skip x samples and sample
                             # Repeat for 21 bits (TBC)
 
+
             #TODO: What happens if it gets stuck in the wrong state?
 
 

protoDshot/__init__.py

@@ -0,0 +1,2 @@
+from .dshot_bits import *
+from .protocols_motor import *