Skip to content

Commit acb6b23

Browse files
hyp0dermik-codehyp0dermik-code
authored
Add unit tests for protocol handler (#6)
* initial unit testing * seperate and test xor and gcr * add packet test and fix regressions of telem handling * fix regressions of telem handling * add custom exception * fix erpm unit tests * tidyup * move add_bit to dshot_common * reimplement telem data display fix state handling add dshot cmd state * add test for telem zero packet * raise error on CRC fail * add .idea to .gitignore
1 parent ff6f5ba commit acb6b23

6 files changed

Lines changed: 198 additions & 103 deletions

File tree

‎.gitignore‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1,2 @@
1-
__pycache__/
1+
__pycache__/
2+
.idea/

‎pd.py‎

Lines changed: 74 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ class State(Enum):
3636
TELEM = 2
3737

3838

39-
class State_Telem(Enum):
39+
class State_Dshot(Enum):
4040
RESET = 0
4141
START = 1
4242
RECV = 2
@@ -94,7 +94,8 @@ def __init__(self):
9494

9595
def reset(self):
9696
self.state = State.CMD
97-
self.state_telem = State_Telem.START
97+
self.state_telem = State_Dshot.START
98+
self.state_dshot = State_Dshot.START
9899

99100
self.samplerate = None
100101
self.inreset = False
@@ -118,7 +119,9 @@ def metadata(self, key, value):
118119
if key == srd.SRD_CONF_SAMPLERATE:
119120
self.samplerate = value
120121

121-
122+
def display_bit(self, bitseq, annot):
123+
self.put(bitseq.ss, bitseq.es, self.out_ann,
124+
[annot, ['%d' % bool(bitseq)]])
122125

123126
def display_dshot(self,dshot):
124127
crc_startsample = dshot.results[12].ss
@@ -139,84 +142,93 @@ def display_dshot(self,dshot):
139142
self.put(crc_startsample, dshot.results[15].es, self.out_ann,
140143
[4, ['CRC INVALID']])
141144

145+
def display_telem(self, telem):
146+
crc_startsample = telem.results[12].ss
147+
self.put(crc_startsample, telem.results[15].es, self.out_ann,
148+
[3, ['Calc CRC: ' + ('%04d' % telem.crc_calc) + ' TXed CRC:' + ('%04d' % telem.crc_recv)]])
149+
if not telem.crc_ok:
150+
self.put(crc_startsample, telem.results[15].es, self.out_ann,
151+
[4, ['CRC INVALID']])
152+
142153

143-
def complete_DshotBit(self, *args):
144-
bitseq = BitDshot(*args)
145154

146-
self.put(bitseq.ss, bitseq.es, self.out_ann,
147-
[0, ['%d' % bool(bitseq)]])
148-
return [bitseq]
149155

150156
def decode(self):
151157
if not self.samplerate:
152158
raise SamplerateError('Cannot decode without samplerate.')
153159

154160
dshot_value = DshotCmd(self.dshot_cfg)
155161
telem_value = DshotTelem(self.dshot_cfg)
156-
results = []
157-
telem = 0b0
158-
tlm_start = 0
159162

160163
#bitseq = BitDshot()
161164
while True:
162165

163166
match self.state:
164167
case State.CMD:
165-
if not self.dshot_cfg.bidirectional:
166-
pins = self.wait([{0: 'r'}, {0: 'f'}, {'skip': self.dshot_cfg.samples_after_motorcmd}])
167-
else:
168-
pins = self.wait([{0: 'f'}, {0: 'r'}, {'skip': self.dshot_cfg.samples_after_motorcmd}])
169-
#TODO: Increase skip to maximum time for effiency
170-
#TODO: Mark any changes in this time as errors? Option to reduce load?
171-
172-
if self.currbit_ss and self.currbit_es and self.matched[2]:
173-
# Assume end of packet if have seen start and end of a potential bit but no further change within 3 periods
174-
# TODO: Confirm wait period this works with spec
175-
176-
args = self.currbit_ss, self.currbit_es, (self.currbit_ss + self.dshot_cfg.samples_pp)
177-
results += self.complete_DshotBit(*args)
178-
self.currbit_ss = None
179-
self.currbit_es = None
180-
#print(results)
181-
# Pass results to decoder
182-
183-
result = dshot_value.handle_bits_dshot(results)
184-
if result:
185-
self.display_dshot(dshot_value)
186-
if result and self.dshot_cfg.bidirectional:
187-
self.state = State.TELEM
188-
189-
results = []
190-
#dshot_value = DshotCmd(self.dshot_cfg)
191-
192-
if self.matched[0] and not self.currbit_ss and not self.currbit_es:
193-
# Start of bit
194-
self.currbit_ss = self.samplenum
195-
elif self.matched[1] and self.currbit_ss and not self.currbit_es:
196-
# End of bit
197-
self.currbit_es = self.samplenum
198-
elif self.matched[0] and self.currbit_es and self.currbit_ss:
199-
# Have complete bit, can handle bit now
200-
args = self.currbit_ss, self.currbit_es, self.samplenum
201-
results += self.complete_DshotBit(*args)
202-
203-
self.currbit_ss = self.samplenum
204-
self.currbit_es = None
168+
match self.state_dshot:
169+
case State_Dshot.RESET:
170+
dshot_value = DshotCmd(self.dshot_cfg)
171+
self.state_dshot = State_Dshot.START
172+
173+
case State_Dshot.START:
174+
if not self.dshot_cfg.bidirectional:
175+
pins = self.wait([{0: 'r'}, {0: 'f'}, {'skip': self.dshot_cfg.samples_after_motorcmd}])
176+
else:
177+
pins = self.wait([{0: 'f'}, {0: 'r'}, {'skip': self.dshot_cfg.samples_after_motorcmd}])
178+
#TODO: Increase skip to maximum time for effiency
179+
#TODO: Mark any changes in this time as errors? Option to reduce load?
180+
181+
if self.currbit_ss and self.currbit_es and self.matched[2]:
182+
# Assume end of packet if have seen start and end of a potential bit but no further change within 3 periods
183+
# TODO: Confirm wait period this works with spec
184+
185+
args = self.currbit_ss, self.currbit_es, (self.currbit_ss + self.dshot_cfg.samples_pp)
186+
curr_bit = BitDshot(*args)
187+
dshot_value.add_bit(curr_bit)
188+
self.display_bit(curr_bit,0)
189+
self.currbit_ss = None
190+
self.currbit_es = None
191+
#print(results)
192+
# Pass results to decoder
193+
194+
result = dshot_value.handle_bits_dshot()
195+
if result:
196+
self.display_dshot(dshot_value)
197+
self.state_dshot = State_Dshot.RESET
198+
if result and self.dshot_cfg.bidirectional:
199+
self.state = State.TELEM
200+
201+
202+
if self.matched[0] and not self.currbit_ss and not self.currbit_es:
203+
# Start of bit
204+
self.currbit_ss = self.samplenum
205+
elif self.matched[1] and self.currbit_ss and not self.currbit_es:
206+
# End of bit
207+
self.currbit_es = self.samplenum
208+
elif self.matched[0] and self.currbit_es and self.currbit_ss:
209+
# Have complete bit, can handle bit now
210+
args = self.currbit_ss, self.currbit_es, self.samplenum
211+
curr_bit = BitDshot(*args)
212+
dshot_value.add_bit(curr_bit)
213+
self.display_bit(curr_bit,0)
214+
215+
self.currbit_ss = self.samplenum
216+
self.currbit_es = None
205217
case State.TELEM:
206218
match self.state_telem:
207-
case State_Telem.RESET:
208-
telem = 0b0
209-
self.state_telem = State_Telem.START
219+
case State_Dshot.RESET:
220+
telem_value = DshotTelem(self.dshot_cfg)
221+
self.state_telem = State_Dshot.START
210222

211-
case State_Telem.START:
223+
case State_Dshot.START:
212224
# First wait for falling edge (idle high)
213225
pins = self.wait([{0: 'f'}])
214226
# Save start pulse
215227
tlm_start = self.samplenum
216228
# Switch to receiving state
217-
self.state_telem = State_Telem.RECV
229+
self.state_telem = State_Dshot.RECV
218230
# TODO: Check if still low after 1/8 bitlength for error det?
219-
case State_Telem.RECV:
231+
case State_Dshot.RECV:
220232
# First conditions skips half bit width and matches low
221233
# Second condition skips half bit width and matches high
222234
pins = self.wait([{0: 'l', 'skip': self.dshot_cfg.telem_baudrate_midpoint},
@@ -225,24 +237,22 @@ def decode(self):
225237
# Append next bit
226238
args = (self.samplenum - self.dshot_cfg.telem_baudrate_midpoint), self.samplenum, (self.samplenum + self.dshot_cfg.telem_baudrate_midpoint)
227239
curr_bit = Bit_DshotTelem(*args,self.matched)
228-
self.put(curr_bit.ss,curr_bit.es,
229-
self.out_ann,
230-
[5, ['%04d' % curr_bit.bit_]])
231-
240+
self.display_bit(curr_bit,5)
232241
telem_value.add_bit(curr_bit)
233242

234243

235244
# Skip half bitwidth to end of bit
236245
pins = self.wait([{'skip': self.dshot_cfg.telem_baudrate_midpoint}])
237246

238-
if telem_value.bits.bit_length() >= 20:
247+
if telem_value.bits.bit_length() >= 20-1:
239248
telem_value.process_telem()
240-
249+
self.display_telem(telem_value)
241250
# Reset
242-
self.state_telem = State_Telem.RESET
251+
self.state_telem = State_Dshot.RESET
243252
# Except Dshot packet next
244253
self.state = State.CMD
245254

255+
246256
# If not mark as error
247257

248258
# Then skip x samples and sample

‎protoDshot/dshot_bits.py‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,18 +16,21 @@ def __init__(self, ss, ts, es):
1616
self.ss, self.ts, self.es = ss, ts, es
1717
self.process_bit()
1818
return
19+
def getBit(self):
20+
if self.bit_ is None:
21+
raise ValueError
22+
return self.bit_
1923
def process_bit(self):
2024

2125
self.period = self.es - self.ss
2226
self.duty = self.ts - self.ss
2327
# Ideal duty for T0H: 33%, T1H: 66%.
2428
self.bit_ = (self.duty / self.period) > 0.5
2529
# TODO: Add tolerance
30+
return self.getBit()
2631

2732
def __bool__(self):
28-
if self.bit_ is None:
29-
raise ValueError
30-
return self.bit_
33+
return bool(self.getBit())
3134
class Bit_DshotTelem(Sequence):
3235
def __init__(self, ss, ts, es, matched):
3336
super().__init__()
@@ -39,6 +42,10 @@ def __init__(self, ss, ts, es, matched):
3942
self.ss, self.ts, self.es = ss, ts, es
4043
self.process_bit(matched)
4144

45+
def getBit(self):
46+
if self.bit_ is None:
47+
raise ValueError
48+
return self.bit_
4249
def process_bit(self,matched):
4350
# Low/High @ given sample
4451
if matched == (True, False):
@@ -49,6 +56,9 @@ def process_bit(self,matched):
4956
if matched == (False, True):
5057
# 1 value
5158
self.bit_ = 1
52-
if self.bit_ is None:
53-
raise ValueError
54-
return self.bit_
59+
return self.getBit()
60+
61+
def __bool__(self):
62+
return bool(self.getBit())
63+
64+

0 commit comments

Comments
 (0)