439 lines
18 KiB
Python
439 lines
18 KiB
Python
"""Wire vectors and DSView API boundary tests; no sigrok installation needed."""
|
|
import binascii
|
|
import importlib
|
|
from pathlib import Path
|
|
import struct
|
|
import sys
|
|
import types
|
|
import tempfile
|
|
import importlib.util
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location('dsview_build', ROOT / 'build.py')
|
|
BUILD = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(BUILD)
|
|
RUNTIME = tempfile.TemporaryDirectory()
|
|
BUILD.package_decoders(Path(RUNTIME.name))
|
|
(Path(RUNTIME.name) / 'decoders/common/__init__.py').write_text('')
|
|
|
|
|
|
class FakeDecoder:
|
|
def has_channel(self, channel):
|
|
return channel < 2
|
|
|
|
def register(self, kind):
|
|
return kind
|
|
|
|
def put(self, ss, es, output, data):
|
|
self.emitted.append((ss, es, output, data))
|
|
|
|
|
|
stub = types.ModuleType('sigrokdecode')
|
|
stub.Decoder = FakeDecoder
|
|
stub.OUTPUT_ANN, stub.OUTPUT_PYTHON, stub.SRD_CONF_SAMPLERATE = 0, 1, 2
|
|
previous = sys.modules.get('sigrokdecode')
|
|
sys.modules['sigrokdecode'] = stub
|
|
sys.path.insert(0, str(Path(RUNTIME.name) / 'decoders'))
|
|
try:
|
|
uart = importlib.import_module('set_uart.pd')
|
|
wire = importlib.import_module('common.setgui_decoders.set_uart')
|
|
can = importlib.import_module('set_can.pd')
|
|
cancore = importlib.import_module('common.setgui_decoders.set_can')
|
|
pm = importlib.import_module('pm35_uart.pd')
|
|
gate = importlib.import_module('gate_driver_timing.pd')
|
|
pair_decoder = importlib.import_module('transistor_pair.pd')
|
|
pmcore = importlib.import_module('common.setgui_decoders.pm35_uart')
|
|
finally:
|
|
sys.path.pop(0)
|
|
if previous is None:
|
|
del sys.modules['sigrokdecode']
|
|
else:
|
|
sys.modules['sigrokdecode'] = previous
|
|
|
|
PING = bytes.fromhex('A5 5A 02 08 01 00 00 00 2A 00 34 12 00 00 33 EC 33 04')
|
|
BRIDGE = bytes.fromhex('AA 55 08 01 01 67 45 23 01 AA BB FE 14')
|
|
|
|
|
|
def crc32_frame(header, payload=b''):
|
|
data = header + payload
|
|
return data + struct.pack('<I', binascii.crc32(data[2:]) & 0xffffffff)
|
|
|
|
|
|
def modbus(raw):
|
|
return raw + struct.pack('<H', wire.crc16(raw, True))
|
|
|
|
|
|
def feed(parser, raw):
|
|
events = []
|
|
for i, byte in enumerate(raw):
|
|
events.extend(parser.feed(byte, i * 10, i * 10 + 9))
|
|
return events
|
|
|
|
|
|
def instance(module, **options):
|
|
decoder = module.Decoder()
|
|
decoder.options = {item['id']: item['default'] for item in decoder.options}
|
|
decoder.options.update(options)
|
|
decoder.emitted = []
|
|
decoder.metadata(2, 1000000)
|
|
decoder.start()
|
|
return decoder
|
|
|
|
|
|
def uart_bytes(decoder, raw, direction=0, base=0):
|
|
for i, byte in enumerate(raw):
|
|
ss = base + i * 100
|
|
decoder.decode(ss, ss+80, ('DATA', direction, (byte, [])))
|
|
decoder.decode(ss, ss+90, ('FRAME', direction, (byte, True)))
|
|
|
|
|
|
def frames(decoder):
|
|
return [item[3][1] for item in decoder.emitted if item[2] == 1]
|
|
|
|
|
|
class WireTests(unittest.TestCase):
|
|
def test_reference_crc(self):
|
|
self.assertEqual(wire.crc16(b'123456789'), 0x29b1)
|
|
self.assertEqual(wire.crc16(b'123456789', True), 0x4b37)
|
|
|
|
def test_documented_ping(self):
|
|
result = wire.parse_frame(PING)
|
|
self.assertEqual((result['destination'], result['sequence']), (42, 0x1234))
|
|
|
|
def test_documented_bridge(self):
|
|
result = wire.parse_frame(BRIDGE)
|
|
self.assertEqual((result['can_id'], result['payload']), (0x1234567, b'\xaa\xbb'))
|
|
|
|
def test_v1_big_endian_header(self):
|
|
frame = crc32_frame(bytes.fromhex('a55a010c12340003'), b'abc')
|
|
result = wire.parse_frame(frame)
|
|
self.assertEqual(result['sequence'], 0x1234)
|
|
self.assertIn('FIRMWARE_DATA', result['summary'])
|
|
|
|
def test_noise_crc_and_resync(self):
|
|
damaged = bytearray(PING)
|
|
damaged[-1] ^= 1
|
|
events = feed(wire.StreamParser(), b'noise\xa5' + damaged + PING + BRIDGE)
|
|
self.assertEqual(sum(e[2] is not None for e in events), 2)
|
|
self.assertTrue(any(e[3] and 'CRC' in e[3] for e in events))
|
|
|
|
def test_bad_version_and_length(self):
|
|
events = feed(wire.StreamParser(), bytes.fromhex('a55aff a55a010100000201') + PING)
|
|
self.assertEqual(sum(e[2] is not None for e in events), 1)
|
|
self.assertGreaterEqual(sum(bool(e[3]) for e in events), 2)
|
|
|
|
def test_max_payload(self):
|
|
raw = crc32_frame(bytes.fromhex('a55a020009000000000001000002'), bytes(range(256))*2)
|
|
events = feed(wire.StreamParser(), raw)
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(len(events[0][2]['payload']), 512)
|
|
|
|
def test_unknown_message_preserved(self):
|
|
raw = crc32_frame(bytes.fromhex('a55a020034120000000001000000'))
|
|
self.assertEqual(wire.parse_frame(raw)['message_type'], 0x1234)
|
|
|
|
def test_uart_directions_are_independent(self):
|
|
decoder = instance(uart)
|
|
for i, byte in enumerate(PING):
|
|
decoder.decode(i*10, i*10+9, ('FRAME', 0, (byte, True)))
|
|
decoder.decode(i*10, i*10+9, ('FRAME', 1, (byte, True)))
|
|
self.assertEqual([f['direction'] for f in frames(decoder)], [0, 1])
|
|
|
|
def test_uart_invalid_last_byte_not_emitted(self):
|
|
decoder = instance(uart)
|
|
uart_bytes(decoder, PING[:-1])
|
|
decoder.decode(2000, 2090, ('FRAME', 0, (PING[-1], False)))
|
|
self.assertFalse(frames(decoder))
|
|
uart_bytes(decoder, PING, base=3000)
|
|
self.assertEqual(len(frames(decoder)), 1)
|
|
|
|
def test_uart_gap_discards_partial(self):
|
|
decoder = instance(uart)
|
|
uart_bytes(decoder, PING[:10])
|
|
uart_bytes(decoder, PING, base=200000)
|
|
self.assertEqual(len(frames(decoder)), 1)
|
|
|
|
|
|
class CanTests(unittest.TestCase):
|
|
def segments(self, packet=PING):
|
|
yield bytes([0x10]) + struct.pack('<H', len(packet)) + packet[:5]
|
|
for i, offset in enumerate(range(5, len(packet), 7), 1):
|
|
yield bytes([0x20 | (i & 15)]) + packet[offset:offset+7]
|
|
|
|
def test_set_reassembly(self):
|
|
decoder = instance(can, protocol='set-v2')
|
|
for i, data in enumerate(self.segments()):
|
|
decoder.decode(i*1000, i*1000+900, ('extended', 0x122a0000, 'data', len(data), list(data)))
|
|
self.assertEqual(frames(decoder)[-1]['raw'], PING)
|
|
|
|
def test_missing_segment(self):
|
|
reader = cancore.Reassembler()
|
|
parts = list(self.segments())
|
|
reader.feed(0x122a0000, parts[0], 0, 9, 0)
|
|
events = reader.feed(0x122a0000, parts[2], 10, 19, 1)
|
|
self.assertIn('sequence', events[-1][3])
|
|
self.assertFalse(reader.pending)
|
|
|
|
def test_timeout_and_new_first(self):
|
|
reader = cancore.Reassembler()
|
|
first = next(self.segments())
|
|
reader.feed(0x122a0000, first, 0, 9, 0)
|
|
events = reader.feed(0x122a0000, first, 20, 29, 500)
|
|
self.assertIn('timeout', events[0][3])
|
|
self.assertEqual(reader.pending[0x122a0000]['start'], 20)
|
|
|
|
def test_id_mismatch(self):
|
|
reader = cancore.Reassembler()
|
|
for i, part in enumerate(self.segments()):
|
|
events = reader.feed(0x122b0000, part, i*10, i*10+9, i)
|
|
self.assertIn('disagree', events[-1][3])
|
|
|
|
def test_sequence_wrap(self):
|
|
packet = crc32_frame(bytes.fromhex('a55a0200010000002a0001008000'), bytes(128))
|
|
reader = cancore.Reassembler()
|
|
for i, part in enumerate(self.segments(packet)):
|
|
events = reader.feed(0x122a0000, part, i*10, i*10+9, i)
|
|
self.assertEqual(events[-1][2]['raw'], packet)
|
|
|
|
def test_interleaved_channels(self):
|
|
reader = cancore.Reassembler()
|
|
complete = []
|
|
for i, part in enumerate(self.segments()):
|
|
for channel in (0, 1):
|
|
complete += reader.feed(0x122a0000 | channel, part, i*10, i*10+9, i)
|
|
self.assertEqual(sum(bool(e[2] and 'raw' in e[2]) for e in complete), 2)
|
|
|
|
def test_gas(self):
|
|
result = cancore.legacy(0x1b530010, b'\x34\x12\xfe\xff')
|
|
self.assertEqual(result['registers'], [(16, 0x1234), (17, 0xfffe)])
|
|
with self.assertRaises(ValueError):
|
|
cancore.legacy(0x1b530010, b'\x00')
|
|
|
|
def test_balsam_mask_and_endianness(self):
|
|
result = cancore.legacy(0xba0010, bytes.fromhex('a01012340000ffff'), 'balsam')
|
|
self.assertEqual(result['registers'], [(16, 0x1234), (18, 0xffff)])
|
|
self.assertIsNone(cancore.legacy(0xba000f, bytes(8), 'balsam'))
|
|
|
|
def test_ignore_standard_and_remote(self):
|
|
decoder = instance(can)
|
|
decoder.decode(0, 9, ('standard', 0x123, 'data', 0, []))
|
|
decoder.decode(10, 19, ('extended', 0x1b530010, 'remote', 0, []))
|
|
self.assertFalse(decoder.emitted)
|
|
|
|
def test_boot_vectors(self):
|
|
self.assertIn('ENTER_BOOT', cancore.legacy(0x13590702, b'')['summary'])
|
|
self.assertIn('offset=0x8', cancore.legacy(0x135b0001, bytes.fromhex('1011121314151617'))['summary'])
|
|
self.assertIn('status=0', cancore.legacy(0x1b5c0702, bytes.fromhex('00ff000000000000'))['summary'])
|
|
|
|
|
|
class PM35Tests(unittest.TestCase):
|
|
def test_single_wire_auto(self):
|
|
reader = pmcore.PM35Parser('auto')
|
|
raw = modbus(bytes.fromhex('100300000002')) + modbus(bytes.fromhex('1003041234ffff'))
|
|
events = feed(reader, raw)
|
|
self.assertEqual(len(events), 2)
|
|
self.assertTrue(all(event[2] for event in events))
|
|
|
|
def test_request_and_response(self):
|
|
request = modbus(bytes.fromhex('100300000002'))
|
|
response = modbus(bytes.fromhex('1003041234ffff'))
|
|
decoder = instance(pm)
|
|
uart_bytes(decoder, request, 1)
|
|
uart_bytes(decoder, response, 0, 10000)
|
|
self.assertEqual(len(frames(decoder)), 2)
|
|
self.assertEqual(frames(decoder)[1]['values'], [0x1234, 65535])
|
|
|
|
def test_128_word_legacy_reply(self):
|
|
events = feed(pmcore.PM35Parser('response'), modbus(b'\x10\x03\x00' + bytes(256)))
|
|
self.assertEqual(len(events[0][2]['values']), 128)
|
|
|
|
def test_write_command(self):
|
|
result = pmcore.parse_pm35(modbus(bytes.fromhex('1006007f8000')), 'request')
|
|
self.assertIn('command_bits=8000', result['summary'])
|
|
|
|
def test_exception(self):
|
|
result = pmcore.parse_pm35(modbus(bytes.fromhex('108302')), 'response')
|
|
self.assertIn('exception', result['summary'])
|
|
|
|
def test_crc_and_range_errors(self):
|
|
with self.assertRaisesRegex(ValueError, 'CRC'):
|
|
pmcore.parse_pm35(bytes.fromhex('1003000000010000'), 'request')
|
|
with self.assertRaisesRegex(ValueError, 'range'):
|
|
pmcore.parse_pm35(modbus(bytes.fromhex('1003007f0002')), 'request')
|
|
|
|
|
|
class GateAdapterTests(unittest.TestCase):
|
|
def test_wait_releases_dsview_shared_pin_tuple(self):
|
|
for channels, pair in (((1,), 0), ((0, 1), 0), ((0, 1), 1)):
|
|
with self.subTest(channels=channels, pair=pair):
|
|
decoder = instance(pair_decoder if pair else gate)
|
|
decoder.has_channel = lambda channel: channel in channels
|
|
shared_pins = tuple([0, 0])
|
|
calls = []
|
|
|
|
def wait(conditions):
|
|
# Native DSView owns one reference and PyTuple_SetItem
|
|
# rejects any reference retained by the previous caller.
|
|
self.assertEqual(sys.getrefcount(shared_pins), 2)
|
|
if len(calls) == 3:
|
|
raise StopIteration
|
|
calls.append(conditions)
|
|
decoder.samplenum = len(calls) * 10
|
|
decoder.matched = 0
|
|
return shared_pins
|
|
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual(len(calls), 3)
|
|
|
|
def test_orphan_filter_boundaries_profiles_and_polarities(self):
|
|
for profile in ('1SP0635', '1SD536F2'):
|
|
for active_high in (True, False):
|
|
with self.subTest(profile=profile, active_high=active_high):
|
|
checker = gate.TimingChecker(1_000_000_000, profile=profile,
|
|
vstat_active_high=active_high, orphan_min_width_ns=50)
|
|
for start, width, expected in ((100, 20, []), (200, 49, []),
|
|
(300, 50, ['orphan']), (400, 51, ['orphan'])):
|
|
self.assertEqual(checker.on_status_edge(start, int(active_high)), [])
|
|
events = checker.on_status_edge(start + width, int(not active_high))
|
|
self.assertEqual([event['kind'] for event in events], expected)
|
|
self.assertIsNone(checker.status_start)
|
|
|
|
def test_orphan_filter_preserves_ack_errors_faults_and_missing_ack(self):
|
|
checker = gate.TimingChecker(1_000_000_000, orphan_min_width_ns=10_000)
|
|
checker.on_control_edge(0, 1)
|
|
self.assertEqual(checker.on_status_edge(250, 1)[0]['kind'], 'delay_ok')
|
|
self.assertEqual(checker.on_status_edge(950, 0)[0]['kind'], 'width_ok')
|
|
checker.on_control_edge(1000, 0)
|
|
checker.on_status_edge(1250, 1)
|
|
self.assertEqual(checker.on_status_edge(1270, 0)[0]['kind'], 'width_fail')
|
|
checker.on_status_edge(2000, 1)
|
|
self.assertEqual(checker.on_status_edge(4000, 0)[0]['kind'], 'fault')
|
|
checker.on_control_edge(5000, 1)
|
|
self.assertEqual(checker.expire(checker.next_deadline())[0]['kind'], 'missing')
|
|
|
|
def test_orphan_filter_default_disabled_and_unknown_duration_retained(self):
|
|
checker = gate.TimingChecker(1_000_000_000)
|
|
checker.on_status_edge(10, 1)
|
|
self.assertEqual(checker.on_status_edge(30, 0)[0]['short'], 'ORPHAN 20.0 ns')
|
|
checker = gate.TimingChecker(1_000_000_000, orphan_min_width_ns=50)
|
|
# A recording starting within a pulse gives no measurable pulse width.
|
|
self.assertEqual(checker.on_status_edge(10, 0)[0]['short'], 'Vstat?')
|
|
|
|
def test_orphan_filter_rejects_invalid_thresholds(self):
|
|
for value in (-1, float('nan'), float('inf')):
|
|
with self.subTest(value=value), self.assertRaisesRegex(ValueError, 'ORPHAN'):
|
|
gate.TimingChecker(100_000_000, orphan_min_width_ns=value)
|
|
|
|
def test_orphan_filter_option_through_native_edge_api(self):
|
|
decoder = instance(gate, orphan_min_width_ns=50)
|
|
decoder.metadata(2, 100_000_000)
|
|
decoder.samplenum = 0
|
|
samples = iter([(10, 1), (12, 0), (20, 1), (25, 0)])
|
|
def wait(conditions):
|
|
decoder.samplenum, level = next(samples)
|
|
decoder.matched = 2
|
|
return 0, level
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual(len(decoder.emitted), 1)
|
|
self.assertEqual(decoder.emitted[0][:2], (20, 25))
|
|
self.assertEqual(decoder.emitted[0][3][1][1], 'ORPHAN 50.0 ns')
|
|
|
|
def test_profiles_through_native_edge_api(self):
|
|
for profile, active, edges, expected in [
|
|
('1SP0635', 'high', [(100, 1, 0, 1), (125, 1, 1, 2), (195, 1, 0, 2)], 700),
|
|
('1SD536F2', 'low', [(20, 1, 1, 1), (58, 1, 0, 2), (148, 1, 1, 2)], 900),
|
|
]:
|
|
with self.subTest(profile=profile):
|
|
decoder = instance(gate, profile=profile, vstat_active=active)
|
|
decoder.metadata(2, 100_000_000)
|
|
decoder.samplenum = 0
|
|
samples = iter(edges)
|
|
def wait(conditions):
|
|
decoder.samplenum, vin, vstat, decoder.matched = next(samples)
|
|
return vin, vstat
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual([item[3][0] for item in decoder.emitted], [0, 1, 2])
|
|
self.assertIn(str(expected), decoder.emitted[-1][3][1][0])
|
|
|
|
def test_driver_has_no_pair_controls(self):
|
|
decoder = instance(gate)
|
|
self.assertNotIn('pair_analysis', decoder.options)
|
|
self.assertNotIn('vin2_mintime_ns', decoder.options)
|
|
self.assertEqual([c['id'] for c in gate.Decoder.optional_channels], ['vin', 'vstat'])
|
|
|
|
def test_pair_decoder_without_vstat(self):
|
|
decoder = instance(pair_decoder, vin2_mintime_ns=200, deadtime_12_ns=50)
|
|
decoder.metadata(2, 1_000_000_000)
|
|
samples = iter([(0, (0, 0)), (10, (1, 0)), (110, (0, 0)),
|
|
(140, (0, 1)), (240, (0, 0)), (280, (1, 0))])
|
|
def wait(conditions):
|
|
decoder.samplenum, pins = next(samples)
|
|
return pins
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
text = ' '.join(item[3][1][0] for item in decoder.emitted)
|
|
self.assertIn('Deadtime Vin1 -> Vin2: 30.0 ns', text)
|
|
self.assertIn('FAIL: Vin2 active', text)
|
|
self.assertIn('duty', text)
|
|
self.assertNotIn('ACK', text)
|
|
self.assertTrue(any(e[3][0] == 3 for e in decoder.emitted))
|
|
decoder.reset()
|
|
self.assertIsNone(decoder.checker)
|
|
|
|
def test_status_only_and_pair_requires_two_inputs(self):
|
|
decoder = instance(gate)
|
|
decoder.has_channel = lambda channel: channel == 1
|
|
samples = iter([(10, 1), (1010, 0)])
|
|
def wait(conditions):
|
|
self.assertEqual(conditions, [{1: 'e'}])
|
|
decoder.samplenum, status = next(samples)
|
|
decoder.matched = 1
|
|
return (255, status, 255)
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual(len(decoder.emitted), 1)
|
|
decoder = instance(pair_decoder)
|
|
decoder.has_channel = lambda channel: channel == 0
|
|
with self.assertRaisesRegex(ValueError, 'Vin1 and Vin2'):
|
|
decoder.decode()
|
|
|
|
def test_missing_ack_timer_emits_error(self):
|
|
decoder = instance(gate)
|
|
decoder.metadata(2, 100_000_000)
|
|
decoder.samplenum = 0
|
|
calls = []
|
|
def wait(conditions):
|
|
calls.append(conditions)
|
|
if len(calls) == 1:
|
|
decoder.samplenum, decoder.matched = 100, 1
|
|
elif len(calls) == 2:
|
|
decoder.samplenum += conditions[2]['skip']
|
|
decoder.matched = 4
|
|
else:
|
|
raise StopIteration
|
|
return 1, 0
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual([item[3][0] for item in decoder.emitted], [0, 4])
|
|
|
|
def test_packaged_core_matches_canonical_templates(self):
|
|
canonical = ROOT.parents[1] / 'python/logic_analyzer/decoders'
|
|
packaged = Path(RUNTIME.name) / 'decoders/common/setgui_decoders'
|
|
for name in ('gate_timing', 'transistor_pair', 'set_uart', 'set_can', 'pm35_uart'):
|
|
self.assertEqual((canonical / (name + '.py')).read_bytes(),
|
|
(packaged / (name + '.py')).read_bytes())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|