jack
323 lines
11 KiB
Python
323 lines
11 KiB
Python
"""Tests for serialcomm/data_parser.py binary parsing pipeline."""
|
|
|
|
import pytest
|
|
from pygui.serialcomm.data_parser import (
|
|
csv_column_count,
|
|
parse_binary_data,
|
|
convert_to_temperatures,
|
|
remove_trailing_zeros,
|
|
save_to_csv,
|
|
_convert_hex_to_temp,
|
|
)
|
|
|
|
# ── csv_column_count ──────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestCsvColumnCount:
|
|
def test_a_family(self):
|
|
assert csv_column_count("A") == 48
|
|
|
|
def test_e_family(self):
|
|
assert csv_column_count("E") == 48
|
|
|
|
def test_p_family(self):
|
|
assert csv_column_count("P") == 48
|
|
|
|
def test_b_family(self):
|
|
assert csv_column_count("B") == 29
|
|
|
|
def test_c_family(self):
|
|
assert csv_column_count("C") == 29
|
|
|
|
def test_d_family(self):
|
|
assert csv_column_count("D") == 29
|
|
|
|
def test_f_family(self):
|
|
assert csv_column_count("F") == 22
|
|
|
|
def test_x_family(self):
|
|
assert csv_column_count("X") == 80
|
|
|
|
def test_unknown_family_returns_zero(self):
|
|
assert csv_column_count("Z") == 0
|
|
|
|
def test_empty_string_returns_zero(self):
|
|
assert csv_column_count("") == 0
|
|
|
|
|
|
# ── Temperature conversion ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestConvertHexToTemp:
|
|
"""Spot-check known hex → temperature values.
|
|
|
|
Reference values derived from the C# ConvertBinaryArrayToDecimal logic.
|
|
"""
|
|
|
|
def test_zero_hex_gives_zero_standard(self):
|
|
# 0x0000 → all bits 0 → temperature 0.0
|
|
result = _convert_hex_to_temp("0000", "B")
|
|
assert result == 0.0
|
|
|
|
def test_zero_hex_gives_zero_aep(self):
|
|
result = _convert_hex_to_temp("0000", "A")
|
|
assert result == 0.0
|
|
|
|
def test_standard_positive_small(self):
|
|
# 0x0050 = 0000 0000 0101 0000 → bits 1-11 = 000000000101 = 4+1 = 5, fraction bits = 00 → 5.0°C
|
|
result = _convert_hex_to_temp("0050", "B")
|
|
assert result == pytest.approx(5.0, abs=0.1)
|
|
|
|
def test_standard_negative(self):
|
|
# Sign bit set → negative temperature
|
|
result = _convert_hex_to_temp("8050", "B")
|
|
assert result < 0
|
|
|
|
def test_aep_positive(self):
|
|
# 0x6400 = 0110 0100 0000 0000
|
|
# sign=0, int bits 1-8 = 11001000 = ...
|
|
# Just check it's in a reasonable range
|
|
result = _convert_hex_to_temp("6400", "A")
|
|
assert -300 < result < 300
|
|
|
|
def test_x_family_uses_aep(self):
|
|
# X uses same AEP formula as A
|
|
result_x = _convert_hex_to_temp("0100", "X")
|
|
result_a = _convert_hex_to_temp("0100", "A")
|
|
assert result_x == result_a
|
|
|
|
def test_unknown_family_falls_back_to_standard(self):
|
|
# Should not raise; falls back to standard formula
|
|
result = _convert_hex_to_temp("0050", "Z")
|
|
assert isinstance(result, float)
|
|
|
|
|
|
# ── Binary parsing ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_p_block(num_blocks: int = 1, value: int = 0x0100) -> bytes:
|
|
"""Build synthetic P-family binary data.
|
|
|
|
Each block is 256 words (512 bytes). Words 0-243 hold `value`,
|
|
words 244-255 are zero (overhead).
|
|
"""
|
|
data = bytearray()
|
|
for _ in range(num_blocks):
|
|
for i in range(256):
|
|
word = value if i < 244 else 0
|
|
data += word.to_bytes(2, byteorder="little")
|
|
return bytes(data)
|
|
|
|
|
|
def _make_x_block(num_blocks: int = 1, value: int = 0x0100) -> bytes:
|
|
"""Build synthetic X-family binary data.
|
|
|
|
Each block is 256 words (512 bytes). Words 0-79 hold `value`,
|
|
words 80-255 are zero (overhead).
|
|
"""
|
|
data = bytearray()
|
|
for _ in range(num_blocks):
|
|
for i in range(256):
|
|
word = value if i < 80 else 0
|
|
data += word.to_bytes(2, byteorder="little")
|
|
return bytes(data)
|
|
|
|
|
|
class TestParseBinaryData:
|
|
def test_p_family_single_block_returns_one_row(self):
|
|
data = _make_p_block(1)
|
|
result = parse_binary_data(data, "P")
|
|
assert result is not None
|
|
assert len(result) == 1
|
|
|
|
def test_p_family_single_block_row_has_244_sensors(self):
|
|
data = _make_p_block(1)
|
|
result = parse_binary_data(data, "P")
|
|
assert result is not None
|
|
assert len(result[0]) == 244
|
|
|
|
def test_p_family_two_blocks_returns_two_rows(self):
|
|
data = _make_p_block(2)
|
|
result = parse_binary_data(data, "P")
|
|
assert result is not None
|
|
assert len(result) == 2
|
|
|
|
def test_p_family_hex_values_are_4_chars(self):
|
|
data = _make_p_block(1)
|
|
result = parse_binary_data(data, "P")
|
|
assert result is not None
|
|
for hex_val in result[0]:
|
|
assert len(hex_val) == 4
|
|
|
|
def test_x_family_single_block_returns_one_row(self):
|
|
data = _make_x_block(1)
|
|
result = parse_binary_data(data, "X")
|
|
assert result is not None
|
|
assert len(result) == 1
|
|
|
|
def test_x_family_single_block_row_has_80_sensors(self):
|
|
data = _make_x_block(1)
|
|
result = parse_binary_data(data, "X")
|
|
assert result is not None
|
|
assert len(result[0]) == 80
|
|
|
|
def test_a_family_uses_p_parser(self):
|
|
# A/B/C/D/E/F all route to _parse_p_binary
|
|
data = _make_p_block(1)
|
|
result = parse_binary_data(data, "A")
|
|
assert result is not None
|
|
assert len(result[0]) == 244
|
|
|
|
def test_empty_bytes_returns_empty_list(self):
|
|
result = parse_binary_data(b"", "P")
|
|
# No full block → no rows
|
|
assert result is not None
|
|
assert result == []
|
|
|
|
|
|
# ── convert_to_temperatures ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestConvertToTemperatures:
|
|
def test_returns_same_shape(self):
|
|
hex_data = [["0000", "0000"], ["0000", "0000"]]
|
|
result = convert_to_temperatures(hex_data, "B")
|
|
assert len(result) == 2
|
|
assert len(result[0]) == 2
|
|
|
|
def test_zero_hex_gives_zero_string(self):
|
|
hex_data = [["0000"]]
|
|
result = convert_to_temperatures(hex_data, "B")
|
|
assert result[0][0] == "0.0"
|
|
|
|
def test_values_are_strings(self):
|
|
hex_data = [["0050"]]
|
|
result = convert_to_temperatures(hex_data, "B")
|
|
assert isinstance(result[0][0], str)
|
|
# Must be parseable as float
|
|
float(result[0][0])
|
|
|
|
def test_p_family_single_block(self):
|
|
data = _make_p_block(1, value=0x0100)
|
|
hex_data = parse_binary_data(data, "P")
|
|
assert hex_data is not None
|
|
result = convert_to_temperatures(hex_data, "P")
|
|
assert len(result) == 1
|
|
assert all(isinstance(v, str) for v in result[0])
|
|
|
|
|
|
# ── remove_trailing_zeros ─────────────────────────────────────────────────────
|
|
|
|
|
|
class TestRemoveTrailingZeros:
|
|
def test_removes_all_zero_last_row(self):
|
|
data = [["25.0", "24.5"], ["0.0", "0.0"]]
|
|
remove_trailing_zeros(data)
|
|
assert len(data) == 1
|
|
|
|
def test_preserves_non_zero_rows(self):
|
|
data = [["25.0", "24.5"], ["23.0", "22.0"]]
|
|
remove_trailing_zeros(data)
|
|
assert len(data) == 2
|
|
|
|
def test_removes_multiple_trailing_zero_rows(self):
|
|
data = [["25.0"], ["0.0"], ["0.0"]]
|
|
remove_trailing_zeros(data)
|
|
assert len(data) == 1
|
|
|
|
def test_all_zero_data_becomes_empty(self):
|
|
data = [["0.0", "0.0"], ["0.0", "0.0"]]
|
|
remove_trailing_zeros(data)
|
|
assert data == []
|
|
|
|
def test_empty_list_stays_empty(self):
|
|
data = []
|
|
remove_trailing_zeros(data)
|
|
assert data == []
|
|
|
|
def test_mixed_keeps_first_nonzero(self):
|
|
data = [["25.0"], ["1.0"], ["0.0"]]
|
|
remove_trailing_zeros(data)
|
|
assert len(data) == 2
|
|
|
|
def test_preserves_negative_temperatures(self):
|
|
data = [["25.0", "24.5"], ["-5.5", "-10.0"]]
|
|
remove_trailing_zeros(data)
|
|
assert len(data) == 2
|
|
|
|
|
|
# ── save_to_csv ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSaveToCsv:
|
|
def test_creates_file(self, tmp_path):
|
|
data = [["25.0", "24.5"], ["23.0", "22.0"]]
|
|
result = save_to_csv(data, "P", "P00001", str(tmp_path))
|
|
assert result is not None
|
|
from pathlib import Path
|
|
|
|
assert Path(result).exists()
|
|
|
|
def test_filename_contains_serial(self, tmp_path):
|
|
data = [["25.0"]]
|
|
result = save_to_csv(data, "P", "P12345", str(tmp_path))
|
|
assert result is not None
|
|
assert "P12345" in result
|
|
|
|
def test_csv_header_wins_when_data_wider_than_display(self, tmp_path):
|
|
# Header count = max(csv_column_count, data width).
|
|
# When data has 60 columns and P display is 48, header is 60 — no truncation.
|
|
data = [[f"{i}.0" for i in range(60)]]
|
|
result = save_to_csv(data, "P", "P00001", str(tmp_path))
|
|
assert result is not None
|
|
headers = open(result).readline().strip().split(",")
|
|
assert len(headers) == 60, f"data width 60 should win over display 48"
|
|
assert headers[0] == "Sensor1"
|
|
assert headers[-1] == "Sensor60"
|
|
|
|
def test_csv_uses_display_count_when_data_is_narrower(self, tmp_path):
|
|
# When data has fewer columns than the display count, pad to display.
|
|
data = [["25.0"]]
|
|
result = save_to_csv(data, "X", "X00001", str(tmp_path))
|
|
assert result is not None
|
|
headers = open(result).readline().strip().split(",")
|
|
# X display = 80, data = 1 → headers = 80
|
|
assert len(headers) == 80
|
|
|
|
def test_csv_row_count_matches_data(self, tmp_path):
|
|
data = [["25.0"], ["24.5"], ["23.0"]]
|
|
result = save_to_csv(data, "P", "P00001", str(tmp_path))
|
|
assert result is not None
|
|
lines = open(result).readlines()
|
|
# 1 header + 3 data rows
|
|
assert len(lines) == 4
|
|
|
|
def test_creates_output_dir_if_missing(self, tmp_path):
|
|
nested = str(tmp_path / "a" / "b" / "c")
|
|
data = [["25.0"]]
|
|
result = save_to_csv(data, "P", "P00001", nested)
|
|
assert result is not None
|
|
from pathlib import Path
|
|
|
|
assert Path(result).exists()
|
|
|
|
def test_returns_none_on_invalid_path(self):
|
|
data = [["25.0"]]
|
|
result = save_to_csv(data, "P", "P00001", "/\x00invalid\x00path")
|
|
assert result is None
|
|
|
|
def test_preserves_all_columns_when_data_wider_than_display(self, tmp_path):
|
|
# Bug fix: when data has more columns than csv_column_count(family),
|
|
# all data columns must be preserved (no silent truncation).
|
|
data = [[f"{i}.0" for i in range(244)]]
|
|
result = save_to_csv(data, "P", "P00001", str(tmp_path))
|
|
assert result is not None
|
|
with open(result) as f:
|
|
lines = f.readlines()
|
|
headers = lines[0].strip().split(",")
|
|
data_row = lines[1].strip().split(",")
|
|
# All 244 columns should be preserved, not truncated to 48
|
|
assert len(headers) == 244, f"expected 244 cols, got {len(headers)}"
|
|
assert len(data_row) == 244, f"expected 244 data cols, got {len(data_row)}"
|
|
assert data_row[243] == "243.0", "last column value must be preserved"
|