Andrey/templates
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Extract reusable device protocols, ports and logic analyzers from SETGUI

Browse Source
This commit is contained in:
andrey
2026-09-23 20:11:37 +03:00
parent 80ba17d77d
commit 795a1279b1
63 changed files with 10181 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

186
python/tests/test_logic_analyzer_analysis.py Normal file
View File

@@ -0,0 +1,186 @@
from array import array
from collections import Counter
from pathlib import Path
import struct
import unittest
from logic_analyzer.files import DigitalChannel, DigitalCapture, read_capture, ImportCancelled
from logic_analyzer.analysis import analyze_capture, can_crc, pulse_measurements, interval_edges
from logic_analyzer.decoders.set_uart import crc16
PING = bytes.fromhex('A5 5A 02 08 01 00 00 00 2A 00 34 12 00 00 33 EC 33 04')
def bits_of(value, count):
return [(value >> bit) & 1 for bit in range(count - 1, -1, -1)]
def capture_bits(bits, rate, invert=False, origin=0):
edges = array('d')
for i in range(1, len(bits)):
if bits[i] != bits[i - 1]:
edges.append(origin + i / rate)
channel = DigitalChannel('bus', bits[0] ^ invert, edges)
return DigitalCapture('test', 'test', [channel], origin, origin + len(bits) / rate)
def uart_capture(data, baud=115200, parity='none', stops=1, invert=False, bad_parity=False, bad_stop=False):
bits = [1] * 5
for byte in data:
bits += [0] + [(byte >> i) & 1 for i in range(8)]
if parity != 'none':
bits += [((sum(bits[-8:]) + (parity == 'odd')) & 1) ^ bad_parity]
bits += [0 if bad_stop else 1] * stops
bits += [1] * 5
return capture_bits(bits, baud, invert)
def can_packet(ident, data, extended=True, remote=False, corrupt_crc=False, ack=True):
if extended:
bits = [0] + bits_of(ident >> 18, 11) + [1, 1] + bits_of(ident & 0x3ffff, 18) + [int(remote), 0, 0]
else:
bits = [0] + bits_of(ident, 11) + [int(remote), 0, 0]
bits += bits_of(len(data), 4)
if not remote:
for byte in data:
bits += bits_of(byte, 8)
bits += bits_of(can_crc(bits) ^ int(corrupt_crc), 15)
stuffed, last, run = [], None, 0
for value in bits:
stuffed.append(value)
run = run + 1 if value == last else 1
last = value
if run == 5:
stuffed.append(1 - value)
last, run = 1 - value, 1
return stuffed + [1, 0 if ack else 1, 1] + [1] * 7
class LogicAnalysisTests(unittest.TestCase):
def test_uart_byte_values_inverted_and_parity(self):
for invert in (False, True):
for parity in ('none', 'even', 'odd'):
cap = uart_capture(b'\x00\xff\xa5\x5a', parity=parity, stops=2, invert=invert)
result = analyze_capture(cap, dict(mode='UART', baudrate=115200, parity=parity, stops=2, inverted=invert))
self.assertEqual([e.text for e in result.events], ['UART 0x00', 'UART 0xFF', 'UART 0xA5', 'UART 0x5A'])
def test_uart_errors_are_not_accepted_as_set_frames(self):
cap = uart_capture(PING, parity='even', bad_parity=True)
result = analyze_capture(cap, dict(mode='SET UART', parity='even'))
self.assertTrue(result.events)
self.assertTrue(all(e.kind == 'error' for e in result.events))
result = analyze_capture(uart_capture(b'\xa5', bad_stop=True), dict(mode='UART'))
self.assertTrue(any('стопового' in e.text for e in result.events))
def test_set_ping_bridge_crc_and_resynchronization(self):
damaged = bytearray(PING)
damaged[-1] ^= 1
data = b'noise' + damaged + PING + bytes.fromhex('AA 55 08 01 01 67 45 23 01 AA BB FE 14')
events = analyze_capture(uart_capture(data), dict(mode='SET UART')).events
self.assertTrue(any('CRC' in e.text and e.kind == 'error' for e in events))
self.assertEqual(sum('SET v2 PING' in e.text for e in events), 1)
self.assertTrue(any('ProtoCAN seq=' in e.text for e in events))
self.assertTrue(any('ProtoCAN GAS' in e.text for e in events))
def test_pm35_request_and_special_response(self):
raw = bytes.fromhex('100300000002')
raw += struct.pack('<H', crc16(raw, True))
events = analyze_capture(uart_capture(raw), dict(mode='PM35 UART', role='request')).events
self.assertEqual(len(events), 1)
self.assertIn('READ R0 count=2', events[0].text)
raw = b'\x10\x03\x00' + bytes(256)
raw += struct.pack('<H', crc16(raw, True))
events = analyze_capture(uart_capture(raw), dict(mode='PM35 UART', role='response')).events
self.assertEqual(len(events), 1)
self.assertIn('128 words', events[0].text)
def test_uart_end_flush_and_gap(self):
result = analyze_capture(uart_capture(PING[:9]), dict(mode='SET UART'))
self.assertTrue(any('Incomplete' in e.text for e in result.events))
cap = uart_capture(PING)
result = analyze_capture(cap, dict(mode='SET UART'), end=cap.end / 2)
self.assertFalse(any(e.kind == 'frame' for e in result.events))
def test_crc15_known_check_value(self):
bits = [bit for byte in b'123456789' for bit in bits_of(byte, 8)]
self.assertEqual(can_crc(bits), 0x059e)
def test_can_standard_extended_remote_and_nack(self):
bits = [1] * 10
bits += can_packet(0x123, b'\x01\x02', extended=False) + [1] * 5
bits += can_packet(0x1b530010, b'\x34\x12\xfe\xff') + [1] * 5
bits += can_packet(0x123, b'\x00\x00', extended=False, remote=True, ack=False) + [1] * 5
result = analyze_capture(capture_bits(bits, 500000), dict(mode='CAN', baudrate=500000))
self.assertEqual(len(result.events), 3, [e.text for e in result.events])
self.assertTrue(all(e.kind == 'can' for e in result.events))
self.assertIn('STD ID=00000123', result.events[0].text)
self.assertIn('EXT ID=1B530010', result.events[1].text)
self.assertEqual(result.events[1].details, '34 12 fe ff')
self.assertIn('RTR NACK', result.events[2].text)
def test_can_crc_error_recovery_and_application(self):
bits = [1] * 10 + can_packet(0x1b530010, b'\x34\x12', corrupt_crc=True) + [1] * 5
bits += can_packet(0x1b530010, b'\x34\x12') + [1] * 5
events = analyze_capture(capture_bits(bits, 1000000), dict(mode='SET CAN')).events
self.assertTrue(any(e.kind == 'error' and 'CRC15' in e.text for e in events))
self.assertEqual(sum('R0010=1234' in e.text for e in events), 1)
def test_can_stuff_error(self):
bits = [1] * 10 + [0] * 20 + [1] * 10
events = analyze_capture(capture_bits(bits, 1000000), dict(mode='CAN')).events
self.assertTrue(any('stuffing' in e.text for e in events))
def test_can_set_reassembly_and_balsam(self):
parts = [bytes([0x10]) + struct.pack('<H', len(PING)) + PING[:5]]
parts += [bytes([0x20 | i]) + PING[offset:offset+7] for i, offset in enumerate(range(5, len(PING), 7), 1)]
bits = [1] * 10
for part in parts:
bits += can_packet(0x122a0000, part) + [1] * 5
events = analyze_capture(capture_bits(bits, 1000000), dict(mode='SET CAN', protocol='set-v2')).events
self.assertTrue(any('SET v2 PING' in e.text for e in events), [e.text for e in events])
bits = [1] * 10 + can_packet(0xba0010, bytes.fromhex('a01012340000ffff')) + [1] * 5
events = analyze_capture(capture_bits(bits, 1000000), dict(mode='SET CAN', protocol='balsam')).events
self.assertTrue(any('R0010=1234 R0012=FFFF' in e.text for e in events))
def test_can_incomplete_end(self):
bits = [1] * 10 + can_packet(0x123, b'\x55', extended=False)[:-8]
events = analyze_capture(capture_bits(bits, 1000000), dict(mode='CAN')).events
self.assertTrue(any('Неполный' in e.text for e in events))
self.assertFalse(any(e.kind == 'can' for e in events))
def test_gate_profiles_and_truncated_window(self):
vin = DigitalChannel('Vin', 0, array('d', [1e-6, 4e-6]))
status = DigitalChannel('Vstat', 0, array('d', [1.25e-6, 1.95e-6]))
cap = DigitalCapture('test', 'test', [vin, status], 0, 6e-6)
events = analyze_capture(cap, dict(mode='1SP0635')).events
self.assertEqual(Counter(e.kind for e in events), dict(control=2, delay_ok=1, width_ok=1, missing=1))
self.assertAlmostEqual(next(e for e in events if e.kind == 'delay_ok').end - 1e-6, 250e-9)
events = analyze_capture(cap, dict(mode='1SP0635'), end=4.2e-6).events
self.assertTrue(any(e.kind == 'incomplete' for e in events))
self.assertFalse(any(e.kind == 'missing' for e in events))
with self.assertRaises(ValueError):
analyze_capture(cap, dict(mode='1SP0635', status_channel=0))
def test_measurements_cycle_and_clipped_pulse(self):
ch = DigitalChannel('D0', 0, array('d', [1e-6, 2e-6, 5e-6, 6e-6]))
measured = pulse_measurements(ch, 1.5e-6, 0, 8e-6)
self.assertAlmostEqual(measured['period'], 4e-6)
self.assertAlmostEqual(measured['duty'], 25)
self.assertEqual(interval_edges(ch, 1e-6, 6e-6), (1, 2))
self.assertNotIn('width', pulse_measurements(ch, 0.5e-6, 0, 8e-6))
def test_result_limit_cancellation_and_progress(self):
cap = uart_capture(bytes(range(100)))
progress = []
result = analyze_capture(cap, dict(mode='UART', max_events=3), progress=progress.append)
self.assertTrue(result.truncated)
self.assertEqual(len(result.events), 3)
self.assertLessEqual(len(progress), 101)
with self.assertRaises(ImportCancelled):
analyze_capture(cap, dict(mode='UART'), cancel=lambda: True)
if __name__ == '__main__':
unittest.main()

115
python/tests/test_logic_analyzer_files.py Normal file
View File

@@ -0,0 +1,115 @@
import csv
import json
from pathlib import Path
import struct
import tempfile
import unittest
import zipfile
from logic_analyzer.files import read_capture, ImportCancelled
def run_bytes(length):
value = length - 1
if value < 64:
return bytes([value])
parts = [value & 127]
value >>= 7
while value >= 64:
parts.append(128 | (value & 127))
value >>= 7
return bytes([64 | value] + list(reversed(parts)))
def sal_binary(version=1, initial=1):
head = b'<SALEAE>' + struct.pack('<IIBdQdBBQ', version, 100, 1, 1000, 0, 0, 0, 0, 2)
for start, end, state, runs in [(0, 100, initial, [2, 98]), (100, 105, initial ^ 1, [3, 2])]:
data = b''.join(run_bytes(n) for n in runs)
if version == 1:
head += struct.pack('<6Q', start, end, end-start, 1000, 1, len(data)) + data
head += struct.pack('<QQQI', 1, 0, 0, state)
else:
head += struct.pack('<QQBBQ', start, end, state, 0, len(data)) + data
return head
class LogicFilesTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
def csv_file(self, text):
path = self.root / 'capture.csv'
path.write_text(text, encoding='utf-8-sig')
return path
def test_csv_preserves_irregular_time_and_channel_names(self):
capture = read_capture(self.csv_file('Time [s],Channel 8,Clock\n127.1,1,0\n127.10000024,0,1\n129,0,1\n'))
self.assertEqual(capture.start, 127.1)
self.assertEqual(capture.end, 129)
self.assertEqual(capture.channels[0].name, 'Channel 8')
self.assertEqual(list(capture.channels[0].edges), [127.10000024])
self.assertEqual(capture.channels[0].level_at(127.1), 1)
self.assertEqual(capture.channels[0].level_at(127.10000024), 0)
def test_dsview_csv_metadata_and_units(self):
capture = read_capture(self.csv_file('; CSV, generated by DSView\n; Sample rate: 100 MHz\nTime(ns), Vin, Vstat\n0,0,1\n240,1,0\n'))
self.assertEqual(capture.format, 'DSLogic CSV')
self.assertAlmostEqual(capture.end, 240e-9)
self.assertEqual(capture.channels[1].name, 'Vstat')
def test_csv_rejects_invalid_data(self):
for rows in ['0,0\n0,1', '1,0\n0,1', 'nan,0', '0,0.5', '0,1,0', '']:
with self.subTest(rows=rows), self.assertRaises(ValueError):
read_capture(self.csv_file('Time [s],D0\n' + rows))
def test_sal_chunks_levels_and_long_runs(self):
for version in (1, 3, 4):
for initial in (0, 1):
path = self.root / 'capture.sal'
with zipfile.ZipFile(path, 'w') as archive:
archive.writestr('meta.json', json.dumps({'data': {'rowsSettings': [
{'channel': {'type': 'Digital', 'deviceChannel': 4}, 'name': 'Clock'}]}}))
archive.writestr('digital-4.bin', sal_binary(version, initial))
capture = read_capture(path)
self.assertEqual(capture.sample_rate, 1000)
self.assertEqual(capture.channels[0].name, 'Clock')
self.assertEqual(capture.channels[0].initial, initial)
self.assertEqual(list(capture.channels[0].edges), [0.002, 0.103])
self.assertEqual(capture.end, 0.105)
def test_sal_rejects_truncated_and_unknown_version(self):
for data in (sal_binary()[:-1], sal_binary(77)):
path = self.root / 'bad.sal'
with zipfile.ZipFile(path, 'w') as archive:
archive.writestr('meta.json', '{}')
archive.writestr('digital-0.bin', data)
with self.assertRaises(ValueError):
read_capture(path)
def test_dsl_sparse_channels_blocks_and_padding(self):
path = self.root / 'capture.dsl'
with zipfile.ZipFile(path, 'w') as archive:
archive.writestr('header', '[version]\nversion=3\n[header]\ndevice mode=0\n'
'samplerate=100 MHz\ntotal samples=10\ntotal blocks=2\n'
'probe3=Vin\nprobe12=Ack\n')
archive.writestr('L-0/0', bytes([0b11110000]))
archive.writestr('L-0/1', bytes([0b11111100]))
archive.writestr('L-1/0', bytes([255]))
archive.writestr('L-1/1', bytes([255]))
capture = read_capture(path)
self.assertEqual([c.name for c in capture.channels], ['Vin', 'Ack'])
self.assertEqual(list(capture.channels[0].edges), [4e-8, 8e-8])
self.assertEqual(capture.channels[1].initial, 1)
self.assertEqual(list(capture.channels[1].edges), [])
self.assertEqual(capture.end, 1e-7)
def test_cancelled_import(self):
with self.assertRaises(ImportCancelled):
read_capture(self.csv_file('Time [s],D0\n0,0'), cancel=lambda: True)
if __name__ == '__main__':
unittest.main()

91
python/tests/test_logic_analyzer_saleae.py Normal file
View File

@@ -0,0 +1,91 @@
from dataclasses import replace
from pathlib import Path
import struct
import tempfile
import unittest
from unittest.mock import MagicMock, patch
from logic_analyzer import saleae
from logic_analyzer.files import ImportCancelled
def binary(path, edges=(), initial=0, start=0., end=1., version=0):
Path(path).write_bytes(struct.pack('<8siiIddQ', b'<SALEAE>', version, 0,
initial, start, end, len(edges)) +
struct.pack('<%dd' % len(edges), *edges))
class SaleaeCoreTests(unittest.TestCase):
def test_constant_channel_tail_and_sparse_channel_numbers(self):
with tempfile.TemporaryDirectory() as directory:
binary(Path(directory) / 'digital_2.bin', [.2, .3])
binary(Path(directory) / 'digital_7.bin', initial=1)
capture = saleae.read_binary(directory, (2, 7))
self.assertEqual([c.name for c in capture.channels], ['D2', 'D7'])
self.assertEqual(capture.end, 1.)
self.assertEqual(capture.channels[0].level_at(.25), 1)
self.assertEqual(capture.channels[0].level_at(.9), 0)
self.assertEqual(capture.channels[1].level_at(.9), 1)
def test_reject_bad_binary_and_cancel(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / 'digital_0.bin'
for kwargs in ({'version': 1}, {'edges': [.4, .3]}, {'edges': [float('nan')]},
{'end': -1}, {'initial': 2}):
binary(path, **kwargs)
with self.assertRaises(ValueError):
saleae.read_binary(directory, (0,))
binary(path, [.2])
with self.assertRaises(ImportCancelled):
saleae.read_binary(directory, (0,), lambda: True)
def test_settings_validation(self):
good = saleae.SaleaeSettings('id')
good.validate()
for kwargs in ({'channels': ()}, {'channels': (1, 1)}, {'channels': (16,)},
{'sample_rate': 0}, {'duration': float('nan')}, {'port': 0},
{'buffer_mb': 1}, {'threshold': 5}, {'device_id': ''}):
with self.assertRaises(ValueError):
replace(good, **kwargs).validate()
def fake_api(self):
api = MagicMock()
manager = api.Manager.connect.return_value.__enter__.return_value
capture = manager.start_capture.return_value
def export(directory, **kwargs):
for channel in kwargs['digital_channels']:
binary(Path(directory) / ('digital_%d.bin' % channel), [.25])
capture.export_raw_data_binary.side_effect = export
return api, manager, capture
def test_stop_exports_and_closes_exactly_once(self):
api, manager, capture = self.fake_api()
settings = saleae.SaleaeSettings('abc', channels=(3, 5), threshold=1.8)
with patch.object(saleae, 'automation_api', return_value=api):
result = saleae.acquire(settings, lambda: True, lambda: False)
self.assertEqual(result.sample_rate, 24000000)
self.assertIn('abc', result.source)
self.assertEqual([c.name for c in result.channels], ['D3', 'D5'])
capture.stop.assert_called_once()
capture.wait.assert_not_called()
capture.close.assert_called_once()
api.LogicDeviceConfiguration.assert_called_once_with(
enabled_digital_channels=[3, 5], digital_sample_rate=24000000, digital_threshold_volts=1.8)
self.assertFalse(Path(capture.export_raw_data_binary.call_args[0][0]).exists())
def test_cancel_stops_without_export_and_failure_closes(self):
api, manager, capture = self.fake_api()
with patch.object(saleae, 'automation_api', return_value=api):
calls = iter((False, True, True))
with self.assertRaises(ImportCancelled):
saleae.acquire(saleae.SaleaeSettings('id'), lambda: False, lambda: next(calls))
capture.stop.assert_called_once()
capture.close.assert_called_once()
capture.export_raw_data_binary.assert_not_called()
capture.reset_mock()
capture.stop.side_effect = RuntimeError('USB disconnected')
with patch.object(saleae, 'automation_api', return_value=api):
with self.assertRaisesRegex(RuntimeError, 'USB'):
saleae.acquire(saleae.SaleaeSettings('id'), lambda: True, lambda: False)
capture.stop.assert_called_once()
capture.close.assert_called_once()

38
python/tests/test_shared_library_boundary.py Normal file
View File

@@ -0,0 +1,38 @@
"""Shared libraries must import without the desktop application or Qt."""
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
class LibraryBoundaryTests(unittest.TestCase):
def test_cores_import_with_gui_and_qt_blocked(self):
root = Path(__file__).resolve().parents[1]
code = '''
import importlib, importlib.abc, pkgutil, sys
class BlockGui(importlib.abc.MetaPathFinder):
def find_spec(self, fullname, path=None, target=None):
if fullname.split('.')[0] in ('gui_desktop', 'PySide6', 'PySide2'):
raise ImportError('Application dependency in shared core: ' + fullname)
sys.meta_path.insert(0, BlockGui())
for name in ('logic_analyzer', 'set_devices'):
package = importlib.import_module(name)
for module in pkgutil.walk_packages(package.__path__, name + '.'):
if '.qt_ports' not in module.name:
importlib.import_module(module.name)
assert 'gui_desktop' not in sys.modules
'''
env = dict(os.environ, PYTHONPATH=str(root))
with tempfile.TemporaryDirectory() as directory:
result = subprocess.run([sys.executable, '-c', code], cwd=directory, env=env,
capture_output=True, text=True, timeout=30)
self.assertEqual(result.returncode, 0, result.stderr)
def test_no_reverse_imports_in_sources(self):
root = Path(__file__).resolve().parents[1]
for name in ('logic_analyzer', 'set_devices', 'altera_logic'):
for path in (root / name).rglob('*.py'):
text = path.read_text(encoding='utf-8')
self.assertNotIn('gui_desktop', text, str(path))