]> sigrok.org Git - libsigrokdecode.git/blobdiff - decoders/onewire/onewire.py
decoder onewire: separated transport layer code from network layer
[libsigrokdecode.git] / decoders / onewire / onewire.py
index 9ece3797813dac7b2f114759b3ad6b72ec5af209..ad0b0a694d870c6eb5759956141175431e1fe004 100644 (file)
 import sigrokdecode as srd
 
 # Annotation feed formats
-ANN_ASCII = 0
-ANN_DEC = 1
-ANN_HEX = 2
-ANN_OCT = 3
-ANN_BITS = 4
+ANN_LINK     = 0
+ANN_NETWORK  = 1
+ANN_TRANSFER = 2
 
 class Decoder(srd.Decoder):
     api_version = 1
@@ -48,11 +46,9 @@ class Decoder(srd.Decoder):
         'overdrive': ['Overdrive', 0],
     }
     annotations = [
-        ['ASCII', 'Data bytes as ASCII characters'],
-        ['Decimal', 'Databytes as decimal, integer values'],
-        ['Hex', 'Data bytes in hex format'],
-        ['Octal', 'Data bytes as octal numbers'],
-        ['Bits', 'Data bytes in bit notation (sequence of 0/1 digits)'],
+        ['Link', 'Link layer events (reset, presence, bit slots)'],
+        ['Network', 'Network layer events (device addressing)'],
+        ['Transfer', 'Transfer layer events'],
     ]
 
     def __init__(self, **kwargs):
@@ -65,16 +61,15 @@ class Decoder(srd.Decoder):
         self.lnk_present = 0
         self.lnk_bit     = 0
         # Network layer variables
-        self.net_state   = 'ROM COMMAND'
-        self.net_event   = 'NONE'
+        self.net_state   = 'IDLE'
         self.net_cnt     = 0
         self.net_search  = "P"
         self.net_data_p  = 0x0
         self.net_data_n  = 0x0
         self.net_data    = 0x0
+        self.net_rom     = 0x0000000000000000
         # Transport layer variables
-        self.trn_state   = 'WAIT FOR EVENT'
-        self.trn_event   = 'NONE'
+        self.trn_state   = 'IDLE'
 
     def start(self, metadata):
         self.samplerate = metadata['samplerate']
@@ -84,7 +79,7 @@ class Decoder(srd.Decoder):
         # The width of the 1-Wire time base (30us) in number of samples.
         # TODO: optimize this value
         self.time_base = float(self.samplerate) * float(0.000030)
-        print ("DEBUG: samplerate = %d, time_base = %d" % (self.samplerate, self.time_base))
+        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_LINK, ['time_base = %d' % self.time_base]])
 
     def report(self):
         pass
@@ -105,8 +100,6 @@ class Decoder(srd.Decoder):
                     self.lnk_fall = self.samplenum
                     # Go to waiting for sample time
                     self.lnk_state = 'WAIT FOR DATA SAMPLE'
-                    self.put(self.lnk_fall, self.samplenum, self.out_ann,
-                             [ANN_DEC, ['LNK: NEGEDGE: ']])
             elif self.lnk_state == 'WAIT FOR DATA SAMPLE':
                 # Data should be sample one 'time unit' after a falling edge
                 if (self.samplenum - self.lnk_fall == 0.5*self.time_base):
@@ -114,8 +107,7 @@ class Decoder(srd.Decoder):
                     self.lnk_event = "DATA BIT"
                     if (self.lnk_bit) :  self.lnk_state = 'WAIT FOR FALLING EDGE'
                     else              :  self.lnk_state = 'WAIT FOR RISING EDGE'
-                    self.put(self.lnk_fall, self.samplenum, self.out_ann,
-                             [ANN_DEC, ['LNK: BIT: ' + str(self.lnk_bit)]])
+                    self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_LINK, ['BIT: %01x' % self.lnk_bit]])
             elif self.lnk_state == 'WAIT FOR RISING EDGE':
                 # The end of a cycle is a rising edge.
                 if (owr == 1):
@@ -126,11 +118,9 @@ class Decoder(srd.Decoder):
                         # Send a reset event to the next protocol layer.
                         self.lnk_event = "RESET"
                         self.lnk_state = "WAIT FOR PRESENCE DETECT"
-                        self.put(self.lnk_fall, self.samplenum, self.out_proto,
-                                 ['RESET'])
-                        self.put(self.lnk_fall, self.samplenum, self.out_ann,
-                                 [ANN_DEC, ['LNK: RESET: ']])
-                        print ("DEBUG: RESET t0=%d t+=%d" % (self.lnk_fall, self.samplenum))
+                        self.put(self.lnk_fall, self.samplenum, self.out_proto, ['RESET'])
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_LINK   , ['RESET']])
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['RESET']])
                         # Reset the timer.
                         self.lnk_fall = self.samplenum
                     # Otherwise this is assumed to be a data bit.
@@ -146,128 +136,147 @@ class Decoder(srd.Decoder):
                     #self.lnk_event   = "PRESENCE DETECT"
                     if (self.lnk_present) :  self.lnk_state = 'WAIT FOR FALLING EDGE'
                     else                  :  self.lnk_state = 'WAIT FOR RISING EDGE'
-                    self.put(self.lnk_fall, self.samplenum, self.out_ann,
-                             [ANN_DEC, ['LNK: PRESENCE: ' + str(self.lnk_present)]])
-                    print ("DEBUG: PRESENCE=%d t0=%d t+=%d" % (self.lnk_present, self.lnk_fall, self.samplenum))
+                    present_str = "False" if self.lnk_present else "True"
+                    self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_LINK   , ['PRESENCE: ' + present_str]])
+                    self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['PRESENCE: ' + present_str]])
             else:
                 raise Exception('Invalid lnk_state: %d' % self.lnk_state)
 
             # Network layer
             
-            # Clear events.
-            self.net_event = "RESET"
             # State machine.
             if (self.lnk_event == "RESET"):
-                self.net_state = "ROM COMMAND"
+                self.net_state = "COMMAND"
                 self.net_search = "P"
                 self.net_cnt    = 0
-            elif (self.lnk_event == "DATA BIT"):
-                if (self.net_state == "ROM COMMAND"):
-                    if (self.collect_data(8)):
-#                        self.put(self.lnk_fall, self.samplenum,
-#                                 self.out_proto, ['LNK: COMMAND', self.net_data])
-                        self.put(self.lnk_fall, self.samplenum, self.out_ann,
-                                 [ANN_DEC, ['NET: ROM COMMAND: 0x' + hex(self.net_data)]])
-                        print ("DEBUG: ROM_COMMAND=0x%02x t0=%d t+=%d" % (self.net_data, self.lnk_fall, self.samplenum))
-                        if   (self.net_data == 0x33):
-                            # READ ROM
-                            self.net_state = "ADDRESS"
-                        elif (self.net_data == 0x0f):
-                            # READ ROM TODO
-                            self.net_state = "ADDRESS"
-                        elif (self.net_data == 0xcc):
-                            # SKIP ROM
-                            self.net_state = "CONTROL COMMAND"
-                        elif (self.net_data == 0x55):
-                            # MATCH ROM
-                            self.net_state = "ADDRESS"
-                        elif (self.net_data == 0xf0):
-                            # SEARCH ROM
-                            self.net_state = "SEARCH"
-                        elif (self.net_data == 0x3c):
-                            # OVERDRIVE SKIP ROM
-                            self.net_state = "CONTROL COMMAND"
-                        elif (self.net_data == 0x69):
-                            # OVERDRIVE MATCH ROM
-                            self.net_state = "ADDRESS"
-                elif (self.net_state == "ADDRESS"):
-                    # family code (1B) + serial number (6B) + CRC (1B)
-                    if (self.collect_data((1+6+1)*8)):
-                        self.net_address = self.net_data & 0xffffffffffffffff
-                        self.net_state = "CONTROL COMMAND"
-                elif (self.net_state == "SEARCH"):
-                    # family code (1B) + serial number (6B) + CRC (1B)
-                    if (self.collect_search((1+6+1)*8)):
-                        self.net_address = self.net_data & 0xffffffffffffffff
-                        self.net_state = "CONTROL COMMAND"
-                elif (self.net_state == "CONTROL COMMAND"):
-                    if (self.collect_data(8)):
-#                        self.put(self.lnk_fall, self.samplenum,
-#                                 self.out_proto, ['LNK: COMMAND', self.net_data])
-                        self.put(self.lnk_fall, self.samplenum, self.out_ann,
-                                 [ANN_DEC, ['NET: FUNCTION COMMAND: 0x' + hex(self.net_data)]])
-                        print ("DEBUG: FUNCTION_COMMAND=0x%02x t0=%d t+=%d" % (self.net_data, self.lnk_fall, self.samplenum))
-                        if   (self.net_data == 0x48):
-                            # COPY SCRATCHPAD
-                            self.net_state = "TODO"
-                        elif (self.net_data == 0x4e):
-                            # WRITE SCRATCHPAD
-                            self.net_state = "TODO"
-                        elif (self.net_data == 0xbe):
-                            # READ SCRATCHPAD
-                            self.net_state = "TODO"
-                        elif (self.net_data == 0xb8):
-                            # RECALL E2
-                            self.net_state = "TODO"
-                        elif (self.net_data == 0xb4):
-                            # READ POWER SUPPLY
-                            self.net_state = "TODO"
-                        else:
-                            # unsupported commands
-                            self.net_state = "UNDEFINED"
-                elif (self.net_state == "UNDEFINED"):
-                    pass
-                else:
-                    raise Exception('Invalid net_state: %s' % self.net_state)
-            elif (self.lnk_event != "NONE"):
-                raise Exception('Invalid lnk_event: %s' % self.lnk_event)
+            elif (self.net_state == "IDLE"):
+                pass
+            elif (self.net_state == "COMMAND"):
+                if (self.collect_data(8)):
+#                    self.put(self.lnk_fall, self.samplenum,
+#                             self.out_proto, ['ROM COMMAND', self.net_data])
+                    self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM COMMAND: 0x%02x' % self.net_data]])
+                    if   (self.net_data == 0x33):
+                        # READ ROM
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM COMMAND: \'READ ROM\'']])
+                        self.net_state = "GET ROM"
+                    elif (self.net_data == 0x0f):
+                        # READ ROM TODO
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM COMMAND: \'READ ROM ???\'']])
+                        self.net_state = "GET ROM"
+                    elif (self.net_data == 0xcc):
+                        # SKIP ROM
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM COMMAND: \'SKIP ROM\'']])
+                        self.net_state = "IDLE"
+                        self.trn_state = "COMMAND"
+                    elif (self.net_data == 0x55):
+                        # MATCH ROM
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM COMMAND: \'MATCH ROM\'']])
+                        self.net_state = "GET ROM"
+                    elif (self.net_data == 0xf0):
+                        # SEARCH ROM
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM COMMAND: \'SEARCH ROM\'']])
+                        self.net_state = "SEARCH ROM"
+                    elif (self.net_data == 0x3c):
+                        # OVERDRIVE SKIP ROM
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM COMMAND: \'OVERDRIVE SKIP ROM\'']])
+                        self.net_state = "IDLE"
+                        self.trn_state = "COMMAND"
+                    elif (self.net_data == 0x69):
+                        # OVERDRIVE MATCH ROM
+                        self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM COMMAND: \'OVERDRIVE MATCH ROM\'']])
+                        self.net_state = "GET ROM"
+            elif (self.net_state == "GET ROM"):
+                # family code (1B) + serial number (6B) + CRC (1B)
+                if (self.collect_data((1+6+1)*8)):
+                    self.net_rom = self.net_data & 0xffffffffffffffff
+                    self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM: 0x%016x' % self.net_rom]])
+                    self.net_state = "IDLE"
+                    self.trn_state = "COMMAND"
+            elif (self.net_state == "SEARCH ROM"):
+                # family code (1B) + serial number (6B) + CRC (1B)
+                if (self.collect_search((1+6+1)*8)):
+                    self.net_rom = self.net_data & 0xffffffffffffffff
+                    self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK, ['ROM: 0x%016x' % self.net_rom]])
+                    self.net_state = "IDLE"
+                    self.trn_state = "COMMAND"
+            else:
+                raise Exception('Invalid net_state: %s' % self.net_state)
 
+            # Transport layer
+            
+            # State machine.
+            if (self.lnk_event == "RESET"):
+                self.trn_state = "IDLE"
+            elif (self.trn_state == "IDLE"):
+                pass
+            elif (self.trn_state == "COMMAND"):
+                if (self.collect_data(8)):
+#                    self.put(self.lnk_fall, self.samplenum, self.out_proto, ['FUNCTION COMMAND', self.net_data])
+                    self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK , ['FUNCTION COMMAND: 0x%02x' % self.net_data]])
+                    self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_TRANSFER, ['FUNCTION COMMAND: 0x%02x' % self.net_data]])
+                    if   (self.net_data == 0x48):
+                        # COPY SCRATCHPAD
+                        self.trn_state = "TODO"
+                    elif (self.net_data == 0x4e):
+                        # WRITE SCRATCHPAD
+                        self.trn_state = "TODO"
+                    elif (self.net_data == 0xbe):
+                        # READ SCRATCHPAD
+                        self.trn_state = "TODO"
+                    elif (self.net_data == 0xb8):
+                        # RECALL E2
+                        self.trn_state = "TODO"
+                    elif (self.net_data == 0xb4):
+                        # READ POWER SUPPLY
+                        self.trn_state = "TODO"
+                    else:
+                        # unsupported commands
+                        self.trn_state = "TODO"
+            elif (self.trn_state == "TODO"):
+#                self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_NETWORK , ['TODO unsupported transfer state: %s' % self.trn_state]])
+#                self.put(self.lnk_fall, self.samplenum, self.out_ann, [ANN_TRANSFER, ['TODO unsupported transfer state: %s' % self.trn_state]])
+                pass
+            else:
+                raise Exception('Invalid trn_state: %s' % self.trn_state)
 
     # Link/Network layer data collector
     def collect_data (self, length):
-        #print ("DEBUG: BIT=%d t0=%d t+=%d" % (self.lnk_bit, self.lnk_fall, self.samplenum))
-        self.net_data = self.net_data & ~(1 << self.net_cnt) | (self.lnk_bit << self.net_cnt)
-        self.net_cnt  = self.net_cnt + 1
-        if (self.net_cnt == length):
-            self.net_data = self.net_data & ((1<<length)-1)
-            self.net_cnt  = 0
-            print ("DEBUG: DATA=0x%0x t0=%d t+=%d" % (self.net_data, self.lnk_fall, self.samplenum))
-            return (1)
+        if (self.lnk_event == "DATA BIT"):
+            #print ("DEBUG: BIT=%d t0=%d t+=%d" % (self.lnk_bit, self.lnk_fall, self.samplenum))
+            self.net_data = self.net_data & ~(1 << self.net_cnt) | (self.lnk_bit << self.net_cnt)
+            self.net_cnt  = self.net_cnt + 1
+            if (self.net_cnt == length):
+                self.net_data = self.net_data & ((1<<length)-1)
+                self.net_cnt  = 0
+                return (1)
+            else:
+                return (0)
         else:
             return (0)
 
     # Link/Network layer search collector
     def collect_search (self, length):
-        #print ("DEBUG: SEARCH=%s BIT=%d t0=%d t+=%d" % (self.net_search, self.lnk_bit, self.lnk_fall, self.samplenum))
-        if   (self.net_search == "P"):
-          self.net_data_p = self.net_data_p & ~(1 << self.net_cnt) | (self.lnk_bit << self.net_cnt)
-          self.net_search = "N"
-        elif (self.net_search == "N"):
-          self.net_data_n = self.net_data_n & ~(1 << self.net_cnt) | (self.lnk_bit << self.net_cnt)
-          self.net_search = "D"
-        elif (self.net_search == "D"):
-          self.net_data   = self.net_data   & ~(1 << self.net_cnt) | (self.lnk_bit << self.net_cnt)
-          self.net_search = "P"
-          self.net_cnt    = self.net_cnt + 1
-        if (self.net_cnt == length):
-            self.net_data_p = self.net_data_p & ((1<<length)-1)
-            self.net_data_n = self.net_data_n & ((1<<length)-1)
-            self.net_data   = self.net_data   & ((1<<length)-1)
-            self.net_search = "P"
-            self.net_cnt    = 0
-            print ("DEBUG: SEARCH_P=0x%0x t0=%d t+=%d" % (self.net_data_p, self.lnk_fall, self.samplenum))
-            print ("DEBUG: SEARCH_N=0x%0x t0=%d t+=%d" % (self.net_data_n, self.lnk_fall, self.samplenum))
-            print ("DEBUG: SEARCH  =0x%0x t0=%d t+=%d" % (self.net_data  , self.lnk_fall, self.samplenum))
-            return (1)
+        if (self.lnk_event == "DATA BIT"):
+            #print ("DEBUG: SEARCH=%s BIT=%d t0=%d t+=%d" % (self.net_search, self.lnk_bit, self.lnk_fall, self.samplenum))
+            if   (self.net_search == "P"):
+              self.net_data_p = self.net_data_p & ~(1 << self.net_cnt) | (self.lnk_bit << self.net_cnt)
+              self.net_search = "N"
+            elif (self.net_search == "N"):
+              self.net_data_n = self.net_data_n & ~(1 << self.net_cnt) | (self.lnk_bit << self.net_cnt)
+              self.net_search = "D"
+            elif (self.net_search == "D"):
+              self.net_data   = self.net_data   & ~(1 << self.net_cnt) | (self.lnk_bit << self.net_cnt)
+              self.net_search = "P"
+              self.net_cnt    = self.net_cnt + 1
+            if (self.net_cnt == length):
+                self.net_data_p = self.net_data_p & ((1<<length)-1)
+                self.net_data_n = self.net_data_n & ((1<<length)-1)
+                self.net_data   = self.net_data   & ((1<<length)-1)
+                self.net_search = "P"
+                self.net_cnt    = 0
+                return (1)
+            else:
+                return (0)
         else:
             return (0)