onewire: added transport layer decoder, it does not do much for now
[libsigrokdecode.git] / 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 # a dictionary of ROM commands and their names, next state
26 command = {0x33: ["READ ROM"              , "GET ROM"   ], 
27            0x0f: ["CONDITIONAL READ ROM"  , "GET ROM"   ],
28            0xcc: ["SKIP ROM"              , "TRANSPORT" ],
29            0x55: ["MATCH ROM"             , "GET ROM"   ],
30            0xf0: ["SEARCH ROM"            , "SEARCH ROM"],
31            0xec: ["CONDITIONAL SEARCH ROM", "SEARCH ROM"],
32            0x3c: ["OVERDRIVE SKIP ROM"    , "TRANSPORT" ],
33            0x6d: ["OVERDRIVE MATCH ROM"   , "GET ROM"   ]}
34
35 class Decoder(srd.Decoder):
36     api_version = 1
37     id = 'onewire_network'
38     name = '1-Wire network layer'
39     longname = '1-Wire serial communication bus'
40     desc = 'Bidirectional, half-duplex, asynchronous serial bus.'
41     license = 'gplv2+'
42     inputs = ['onewire_link']
43     outputs = ['onewire_network']
44     probes = []
45     optional_probes = []
46     options = {}
47     annotations = [
48         ['Network', 'Network layer events (device addressing)'],
49     ]
50
51     def __init__(self, **kwargs):
52         # Event timing variables
53         self.net_beg = 0
54         self.net_end = 0
55         # Network layer variables
56         self.state   = 'COMMAND'
57         self.bit_cnt = 0
58         self.search  = "P"
59         self.data_p  = 0x0
60         self.data_n  = 0x0
61         self.data    = 0x0
62         self.net_rom = 0x0000000000000000
63
64     def start(self, metadata):
65         self.out_proto = self.add(srd.OUTPUT_PROTO, 'onewire_network')
66         self.out_ann   = self.add(srd.OUTPUT_ANN  , 'onewire_network')
67
68     def report(self):
69         pass
70
71     def decode(self, ss, es, data):
72         [code, val] = data
73
74         # State machine.
75         if (code == "RESET/PRESENCE"):
76             self.search = "P"
77             self.bit_cnt = 0
78             self.put(ss, es, self.out_ann, [0, ['RESET/PRESENCE: %s' % ('True' if val else 'False')]])
79             self.put(ss, es, self.out_proto, ['RESET/PRESENCE', val])
80             self.state = "COMMAND"
81         elif (code == "BIT"):
82             if (self.state == "COMMAND"):
83                 # Receiving and decoding a ROM command
84                 if (self.onewire_collect(8, val, ss, es)):
85                     if (self.data in command):
86                         self.put(self.net_beg, self.net_end, self.out_ann, [0,
87                           ['ROM COMMAND: 0x%02x \'%s\'' % (self.data, command[self.data][0])]])
88                         self.state = command[self.data][1]
89                     else:
90                         self.put(self.net_beg, self.net_end, self.out_ann, [0,
91                           ['ROM COMMAND: 0x%02x \'%s\'' % (self.data, 'UNRECOGNIZED')]])
92                         self.state = "COMMAND ERROR"
93             elif (self.state == "GET ROM"):
94                 # A 64 bit device address is selected
95                 # family code (1B) + serial number (6B) + CRC (1B)
96                 if (self.onewire_collect(64, val, ss, es)):
97                     self.net_rom = self.data & 0xffffffffffffffff
98                     self.put(self.net_beg, self.net_end, self.out_ann, [0, ['ROM: 0x%016x' % self.net_rom]])
99                     self.put(self.net_beg, self.net_end, self.out_proto, ['ROM', self.net_rom])
100                     self.state = "TRANSPORT"
101             elif (self.state == "SEARCH ROM"):
102                 # A 64 bit device address is searched for
103                 # family code (1B) + serial number (6B) + CRC (1B)
104                 if (self.onewire_search(64, val, ss, es)):
105                     self.net_rom = self.data & 0xffffffffffffffff
106                     self.put(self.net_beg, self.net_end, self.out_ann, [0, ['ROM: 0x%016x' % self.net_rom]])
107                     self.put(self.net_beg, self.net_end, self.out_proto, ['ROM', self.net_rom])
108                     self.state = "TRANSPORT"
109             elif (self.state == "TRANSPORT"):
110                 # The transport layer is handled in byte sized units
111                 if (self.onewire_collect(8, val, ss, es)):
112                     self.put(self.net_beg, self.net_end, self.out_ann, [0, ['DATA: 0x%02x' % self.data]])
113                     self.put(self.net_beg, self.net_end, self.out_proto, ['DATA', self.data])
114             elif (self.state == "COMMAND ERROR"):
115                 # Since the command is not recognized, print raw data
116                 if (self.onewire_collect(8, val, ss, es)):
117                     self.put(self.net_beg, self.net_end, self.out_ann, [0, ['ROM ERROR DATA: 0x%02x' % self.data]])
118             else:
119                 raise Exception('Invalid state: %s' % self.state)
120
121
122     # Link/Network layer data collector
123     def onewire_collect (self, length, val, ss, es):
124         # Storing the sampe this sequence begins with
125         if (self.bit_cnt == 1):
126             self.net_beg = ss
127         self.data = self.data & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
128         self.bit_cnt  = self.bit_cnt + 1
129         # Storing the sampe this sequence ends with
130         # In case the full length of the sequence is received, return 1
131         if (self.bit_cnt == length):
132             self.net_end  = es
133             self.data = self.data & ((1<<length)-1)
134             self.bit_cnt  = 0
135             return (1)
136         else:
137             return (0)
138
139     # Link/Network layer search collector
140     def onewire_search (self, length, val, ss, es):
141         # Storing the sampe this sequence begins with
142         if ((self.bit_cnt == 0) and (self.search == "P")):
143             self.net_beg = ss
144         # Master receives an original address bit
145         if   (self.search == "P"):
146           self.data_p = self.data_p & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
147           self.search = "N"
148         # Master receives a complemented address bit
149         elif (self.search == "N"):
150           self.data_n = self.data_n & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
151           self.search = "D"
152         # Master transmits an address bit
153         elif (self.search == "D"):
154           self.data   = self.data   & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
155           self.search = "P"
156           self.bit_cnt    = self.bit_cnt + 1
157         # Storing the sampe this sequence ends with
158         # In case the full length of the sequence is received, return 1
159         if (self.bit_cnt == length):
160             self.net_end = es
161             self.data_p = self.data_p & ((1<<length)-1)
162             self.data_n = self.data_n & ((1<<length)-1)
163             self.data   = self.data   & ((1<<length)-1)
164             self.search = "P"
165             self.bit_cnt    = 0
166             return (1)
167         else:
168             return (0)