2 ## This file is part of the sigrok project.
4 ## Copyright (C) 2012 Iztok Jeras <iztok.jeras@gmail.com>
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.
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.
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
21 # 1-Wire protocol decoder
23 import sigrokdecode as srd
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" ]}
35 class Decoder(srd.Decoder):
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.'
42 inputs = ['onewire_link']
43 outputs = ['onewire_network']
48 ['Network', 'Network layer events (device addressing)'],
51 def __init__(self, **kwargs):
52 # Event timing variables
55 # Network layer variables
56 self.state = 'COMMAND'
62 self.net_rom = 0x0000000000000000
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')
71 def decode(self, ss, es, data):
75 if (code == "RESET/PRESENCE"):
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"
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]
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]])
119 raise Exception('Invalid state: %s' % self.state)
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):
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):
133 self.data = self.data & ((1<<length)-1)
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")):
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)
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)
152 # Master transmits an address bit
153 elif (self.search == "D"):
154 self.data = self.data & ~(1 << self.bit_cnt) | (val << self.bit_cnt)
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):
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)