1"""Protocol checks run without serial hardware or third-party packages."""23import math4from pathlib import Path5import struct6import sys7import unittest89sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "software" / "examples"))1011from hall_protocol import (CMD_GET_DATA, TYPE_ENCODER, TYPE_POSITION, Parser,12 ProtocolError, build_frame, crc16, parse_info, parse_sample)13from read_sensor import transact141516class WireProtocolTests(unittest.TestCase):17 def test_known_crc_vectors_and_request(self):18 self.assertEqual(crc16(b"123456789"), 0x4B37)19 self.assertEqual(crc16(b""), 0xFFFF)20 # Cross-checked against original host crc16; coverage is DST through LEN.21 self.assertEqual(build_frame(1, 0, 1, CMD_GET_DATA, 1).hex(" "),22 "a5 5a 01 00 01 10 01 00 00 63")2324 def test_every_split_boundary_and_concatenation(self):25 wire = build_frame(1, 0, 1, CMD_GET_DATA, 1)26 for cut in range(1, len(wire)):27 parser = Parser()28 self.assertEqual(parser.feed(wire[:cut]), [])29 frames = parser.feed(wire[cut:] + wire)30 self.assertEqual(len(frames), 2)31 self.assertEqual(frames[0].seq, 1)32 parser = Parser()33 frames = []34 for byte in b"noise\xa5" + wire + wire:35 frames.extend(parser.feed(bytes((byte,))))36 self.assertEqual(len(frames), 2)3738 def test_bad_crc_is_rejected_and_next_frame_recovers(self):39 good = build_frame(1, 0, 1, CMD_GET_DATA, 1)40 bad = bytearray(good)41 bad[-1] ^= 0x8042 parser = Parser()43 self.assertEqual(parser.feed(bytes(bad)), [])44 self.assertEqual(len(parser.feed(good)), 1)45 self.assertEqual(parser.crc_errors, 1)4647 def test_overlength_frame_recovers_and_builder_rejects(self):48 good = build_frame(1, 0, 1, CMD_GET_DATA, 1)49 parser = Parser()50 frames = parser.feed(b"\xa5\x5a\x01\x00\x01\x10\x01\x21" + good)51 self.assertEqual(len(frames), 1)52 self.assertEqual(parser.length_errors, 1)53 with self.assertRaises(ValueError):54 build_frame(1, 0, 1, 0x10, 1, bytes(33))5556 def test_profile_changes_units_and_preserves_signed_line_value(self):57 angle = b"\x00" + struct.pack("<ihHHHB", 9000, 1000, 2048, 2000, 2100, 2)58 sample = parse_sample(angle, TYPE_ENCODER, host_rx_unix_s=100.0, host_rx_monotonic_s=5.0)59 self.assertAlmostEqual(sample.si_value, math.pi / 2)60 self.assertEqual(sample.to_dict()["si_unit"], "rad")61 self.assertEqual(sample.to_dict()["raw_unit"], "0.01_deg")62 self.assertEqual(sample.host_rx_unix_s, 100.0)63 self.assertIn("not_hardware_sample_time", sample.to_dict()["timestamp_kind"])64 line = b"\x00" + struct.pack("<ihHHHB", -1250, 1000, 2048, 2000, 2100, 8)65 sample = parse_sample(line, TYPE_POSITION)66 self.assertEqual(sample.raw_value, -1250)67 self.assertAlmostEqual(sample.si_value, -0.0125)68 self.assertEqual(sample.to_dict()["si_unit"], "m")69 self.assertEqual(sample.to_dict()["flag_names"], ["out_of_array"])7071 def test_response_status_report_layout_and_unknown_profile(self):72 raw = struct.pack("<ihHHHB", 1234, 80, 1, 2, 3, 1)73 self.assertEqual(parse_sample(raw, TYPE_ENCODER, with_status=False).raw_value, 1234)74 with self.assertRaisesRegex(ProtocolError, "status 0x04"):75 parse_sample(b"\x04", TYPE_ENCODER)76 with self.assertRaises(ProtocolError):77 parse_sample(raw, TYPE_ENCODER)78 with self.assertRaises(ProtocolError):79 parse_sample(b"\x00" + raw, 0xFF)8081 def test_info_uses_signed_zero_only_for_line_profile(self):82 info = bytes((0, 2, 1, 0, 4)) + struct.pack("<HHBBh", 20, 0x0101, 1, 2, -120)83 parsed = parse_info(info)84 self.assertEqual(parsed["dev_type"], TYPE_POSITION)85 self.assertEqual(parsed["report_ms"], 20)86 self.assertEqual(parsed["zero_raw"], -120)8788 def test_transaction_ignores_other_device_stale_seq_and_reports(self):89 sample = b"\x00" + struct.pack("<ihHHHB", -1250, 1000, 2048, 2200, 2500, 2)90 expected = build_frame(0, 1, 2, 0x90, 7, sample)91 unrelated = b"".join((92 build_frame(0, 2, 2, 0x90, 7, sample), # Another device.93 build_frame(0, 1, 2, 0x90, 6, sample), # Old transaction.94 build_frame(0, 1, 1, 0x90, 7, sample), # Wrong profile/units.95 build_frame(3, 1, 2, 0x90, 7, sample), # Another destination.96 build_frame(0, 1, 2, 0x30, 7, sample[1:]), # An unsolicited report.97 ))9899 class FakePort:100 def __init__(self, incoming):101 self.incoming = bytearray(incoming)102 self.written = b""103104 @property105 def in_waiting(self):106 return len(self.incoming)107108 def write(self, data):109 self.written += data110 return len(data)111112 def flush(self):113 pass114115 def read(self, size):116 data = bytes(self.incoming[:size])117 del self.incoming[:size]118 return data119120 port = FakePort(unrelated + expected)121 frame = transact(port, Parser(), 1, 2, 0x10, 7, 0.1)122 self.assertEqual(frame.payload, sample)123 self.assertEqual(port.written, build_frame(1, 0, 2, 0x10, 7))124 with self.assertRaises(TimeoutError):125 transact(FakePort(b""), Parser(), 1, 2, 0x10, 7, 0.001)126127 class ShortWritePort(FakePort):128 def write(self, data):129 return 1130131 with self.assertRaisesRegex(ProtocolError, "complete request"):132 transact(ShortWritePort(b""), Parser(), 1, 2, 0x10, 7, 0.1)133134 def test_incomplete_false_header_recovers_after_idle_gap(self):135 good = build_frame(1, 0, 1, CMD_GET_DATA, 1)136 false_header = b"\xa5\x5a\x01\x00\x01\x10\x01\x20"137 parser = Parser()138 self.assertEqual(parser.feed(false_header + good, now=1.0), [])139 self.assertEqual(parser.feed(b"", now=1.01), [])140 self.assertEqual(len(parser.feed(b"", now=1.03)), 1)141 self.assertEqual(parser.partial_timeouts, 1)142143 def test_invalid_information_modes_and_calibration_bits_are_rejected(self):144 info = bytes((0, 2, 1, 0, 4)) + struct.pack("<HHBBh", 20, 0x0101, 1, 2, -120)145 for index, value in ((2, 0), (3, 2), (4, 8), (5, 1), (9, 0), (10, 0x82)):146 bad = bytearray(info)147 bad[index] = value148 with self.subTest(index=index, value=value), self.assertRaises(ProtocolError):149 parse_info(bytes(bad))150151 def test_bad_codes_are_rejected_and_line_zero_offset_is_not_clamped(self):152 for value, amp, adc, flags in ((36000, 80, 1000, 2), (-1, 80, 1000, 2),153 (1234, -1, 1000, 2), (1234, 80, 4096, 2),154 (1234, 80, 1000, 8)):155 raw = b"\x00" + struct.pack("<ihHHHB", value, amp, adc, 2, 3, flags)156 with self.subTest(value=value, amp=amp, adc=adc, flags=flags), self.assertRaises(ProtocolError):157 parse_sample(raw, TYPE_ENCODER)158 raw = b"\x00" + struct.pack("<ihHHHB", -2000, 80, 1000, 2, 3, 2)159 sample = parse_sample(raw, TYPE_POSITION)160 self.assertEqual(sample.raw_value, -2000)161 self.assertAlmostEqual(sample.si_value, -0.02)162163 def test_warning_flags_are_exposed_without_claiming_control_validity(self):164 raw = b"\x00" + struct.pack("<ihHHHB", 10, 10, 1, 2, 3, 0x0D)165 row = parse_sample(raw, TYPE_POSITION).to_dict()166 self.assertFalse(row["calibration_valid_flag"])167 self.assertEqual(row["measurement_warnings"],168 ["weak_field", "calibrating", "outside_array_or_interpolation",169 "calibration_not_recorded"])170171172if __name__ == "__main__":173 unittest.main()
元驭机器人Hall 传感器教学下载原文件