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