decoders/onewire_network/onewire_network.py

   1 ##
   2 ## This file is part of the sigrok project.
   3 ##
   4 ## Copyright (C) 2012 Iztok Jeras <iztok.jeras@gmail.com>
   5 ##
   6 ## This program is free software; you can redistribute it and/or modify
   7 ## it under the terms of the GNU General Public License as published by
   8 ## the Free Software Foundation; either version 2 of the License, or
   9 ## (at your option) any later version.
  10 ##
  11 ## This program is distributed in the hope that it will be useful,
  12 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 ## GNU General Public License for more details.
  15 ##
  16 ## You should have received a copy of the GNU General Public License
  17 ## along with this program; if not, write to the Free Software
  18 ## Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  19 ##
  20
  21 # 1-Wire protocol decoder
  22
  23 import sigrokdecode as srd
  24
  25 # Annotation feed formats
  26 ANN_NETWORK   = 0
  27 ANN_TRANSPORT = 1
  28
  29 # a dictionary of ROM commands and their names
  30 rom_command = {0x33: "READ ROM",
  31                0x0f: "CONDITIONAL READ ROM",
  32                0xcc: "SKIP ROM",
  33                0x55: "MATCH ROM",
  34                0xf0: "SEARCH ROM",
  35                0xec: "CONDITIONAL SEARCH ROM",
  36                0x3c: "OVERDRIVE SKIP ROM",
  37                0x6d: "OVERDRIVE MATCH ROM"}
  38
  39 class Decoder(srd.Decoder):
  40     api_version = 1
  41     id = 'onewire_network'
  42     name = '1-Wire network layer'
  43     longname = '1-Wire serial communication bus'
  44     desc = 'Bidirectional, half-duplex, asynchronous serial bus.'
  45     license = 'gplv2+'
  46     inputs = ['onewire_link']
  47     outputs = ['onewire_network']
  48     probes = []
  49     optional_probes = []
  50     options = {}
  51     annotations = [
  52         ['Network', 'Network layer events (device addressing)'],
  53         ['Transport', 'Transport layer events'],
  54     ]
  55
  56     def __init__(self, **kwargs):
  57         # Event timing variables
  58         self.net_beg = 0
  59         self.net_end = 0
  60         # Network layer variables
  61         self.state   = 'COMMAND'
  62         self.bit_cnt = 0
  63         self.search  = "P"
  64         self.data_p  = 0x0
  65         self.data_n  = 0x0
  66         self.data    = 0x0
  67         self.net_rom = 0x0000000000000000
  68
  69     def start(self, metadata):
  70         self.out_proto = self.add(srd.OUTPUT_PROTO, 'onewire_network')
  71         self.out_ann   = self.add(srd.OUTPUT_ANN  , 'onewire_network')
  72
  73     def report(self):
  74         pass
  75
  76     def decode(self, ss, es, data):
  77         [code, val] = data
  78
  79         # State machine.
  80         if (code == "RESET/PRESENCE"):
  81             self.state = "COMMAND"
  82             self.search = "P"
  83             self.bit_cnt = 0
  84         elif (code == "BIT"):
  85             if (self.state == "COMMAND"):
  86                 # Receiving and decoding a ROM command
  87                 if (self.onewire_collect(8, val, ss, es)):
  88                     self.put(self.net_beg, self.net_end, self.out_ann, [ANN_NETWORK,
  89                       ['ROM COMMAND: 0x%02x \'%s\'' % (self.data, rom_command[self.data])]])
  90                     if   (self.data == 0x33):  # READ ROM
  91                         self.state = "GET ROM"
  92                     elif (self.data == 0x0f):  # CONDITIONAL READ ROM
  93                         self.state = "GET ROM"
  94                     elif (self.data == 0xcc):  # SKIP ROM
  95                         self.state = "TRANSPORT"
  96                     elif (self.data == 0x55):  # MATCH ROM
  97                         self.state = "GET ROM"
  98                     elif (self.data == 0xf0):  # SEARCH ROM
  99                         self.state = "SEARCH ROM"
 100                     elif (self.data == 0xec):  # CONDITIONAL SEARCH ROM
 101                         self.state = "SEARCH ROM"
 102                     elif (self.data == 0x3c):  # OVERDRIVE SKIP ROM
 103                         self.state = "TRANSPORT"
 104                     elif (self.data == 0x69):  # OVERDRIVE MATCH ROM
 105                         self.state = "GET ROM"
 106             elif (self.state == "GET ROM"):
 107                 # A 64 bit device address is selected
 108                 # family code (1B) + serial number (6B) + CRC (1B)
 109                 if (self.onewire_collect(64, val, ss, es)):
 110                     self.net_rom = self.data & 0xffffffffffffffff
 111                     self.put(self.net_beg, self.net_end, self.out_ann, [ANN_NETWORK, ['ROM: 0x%016x' % self.net_rom]])
 112                     self.state = "TRANSPORT"
 113             elif (self.state == "SEARCH ROM"):
 114                 # A 64 bit device address is searched for
 115                 # family code (1B) + serial number (6B) + CRC (1B)
 116                 if (self.onewire_search(64, val, ss, es)):
 117                     self.net_rom = self.data & 0xffffffffffffffff
 118                     self.put(self.net_beg, self.net_end, self.out_ann, [ANN_NETWORK, ['ROM: 0x%016x' % self.net_rom]])
 119                     self.state = "TRANSPORT"
 120             elif (self.state == "TRANSPORT"):
 121                 # The transport layer is handled in byte sized units
 122                 if (self.onewire_collect(8, val, ss, es)):
 123                     self.put(self.net_beg, self.net_end, self.out_ann, [ANN_NETWORK  , ['TRANSPORT: 0x%02x' % self.data]])
 124                     self.put(self.net_beg, self.net_end, self.out_ann, [ANN_TRANSPORT, ['TRANSPORT: 0x%02x' % self.data]])
 125                     self.put(self.net_beg, self.net_end, self.out_proto, ['transfer', self.data])
 126                     # TODO: Sending translort layer data to 1-Wire device models
 127             else:
 128                 raise Exception('Invalid state: %s' % self.state)
 129
 130
 131     # Link/Network layer data collector
 132     def onewire_collect (self, length, val, ss, es):
 133         # Storing the sampe this sequence begins with
 134         if (self.bit_cnt == 1):
 135             self.net_beg = ss
 136         self.data = self.data & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
 137         self.bit_cnt  = self.bit_cnt + 1
 138         # Storing the sampe this sequence ends with
 139         # In case the full length of the sequence is received, return 1
 140         if (self.bit_cnt == length):
 141             self.net_end  = es
 142             self.data = self.data & ((1<<length)-1)
 143             self.bit_cnt  = 0
 144             return (1)
 145         else:
 146             return (0)
 147
 148     # Link/Network layer search collector
 149     def onewire_search (self, length, val, ss, es):
 150         # Storing the sampe this sequence begins with
 151         if ((self.bit_cnt == 0) and (self.search == "P")):
 152             self.net_beg = ss
 153         # Master receives an original address bit
 154         if   (self.search == "P"):
 155           self.data_p = self.data_p & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
 156           self.search = "N"
 157         # Master receives a complemented address bit
 158         elif (self.search == "N"):
 159           self.data_n = self.data_n & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
 160           self.search = "D"
 161         # Master transmits an address bit
 162         elif (self.search == "D"):
 163           self.data   = self.data   & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
 164           self.search = "P"
 165           self.bit_cnt    = self.bit_cnt + 1
 166         # Storing the sampe this sequence ends with
 167         # In case the full length of the sequence is received, return 1
 168         if (self.bit_cnt == length):
 169             self.net_end = es
 170             self.data_p = self.data_p & ((1<<length)-1)
 171             self.data_n = self.data_n & ((1<<length)-1)
 172             self.data   = self.data   & ((1<<length)-1)
 173             self.search = "P"
 174             self.bit_cnt    = 0
 175             return (1)
 176         else:
 177             return (0)