diff --git a/src/pyfaradaycup/decommutator/ccsds_reader_pipeline.py b/src/pyfaradaycup/decommutator/ccsds_reader_pipeline.py index da21450..b70b1c6 100644 --- a/src/pyfaradaycup/decommutator/ccsds_reader_pipeline.py +++ b/src/pyfaradaycup/decommutator/ccsds_reader_pipeline.py @@ -19,6 +19,7 @@ import struct import sys import time +import warnings from typing import TYPE_CHECKING import dateutil.parser @@ -330,10 +331,10 @@ def read_file_sc( # ruff:ignore[C901, PLR0912, PLR0915] # and thus which SC_HK.blk file to use # we'll assume the first bytes in the file are a header try: - if ptp: # ruff:ignore[SIM108] - cchead = _parse_ccsds_head(bytestr[17:]) + if ptp: + cchead = _parse_ccsds_head(bytestr[17:27]) else: - cchead = _parse_ccsds_head(bytestr) + cchead = _parse_ccsds_head(bytestr[:10]) if ( (cchead["CCSDS_Version"] != 0) | (cchead["CCSDS_PacketType"] != 0) @@ -576,6 +577,11 @@ def _parse_ccsds_head(bytestr: bytes) -> dict[str, int]: ValueError If ``bytestr`` is shorter than 10 bytes. + Warns + ----- + UserWarning + If ``bytestr`` is longer than 10 bytes. + Notes ----- The first 6 bytes are the standard CCSDS primary header. The next @@ -587,6 +593,16 @@ def _parse_ccsds_head(bytestr: bytes) -> dict[str, int]: exp_length = 10 if len(bytearr) < exp_length: raise ValueError("CCSDS header is not as long as expected") # ruff:ignore[EM101, TRY003] + # For now, issue a warning when the bytearr is too long. + # If bytearr being too long turns out to be normal and + # expected behavior, we can remove this warning later, + # or set it to the debug level of verbosity with a logger. + if len(bytearr) > exp_length: + warnings.warn( + f"CCSDS header is {len(bytearr)} bytes long; only the first " + f"{exp_length} bytes will be decoded", + stacklevel=2, + ) head = {} head["CCSDS_Version"] = bytearr[0] >> 5 diff --git a/tests/decommutator/test_ccsds_reader_pipeline.py b/tests/decommutator/test_ccsds_reader_pipeline.py index 65657aa..f6858ce 100644 --- a/tests/decommutator/test_ccsds_reader_pipeline.py +++ b/tests/decommutator/test_ccsds_reader_pipeline.py @@ -119,16 +119,19 @@ def test_parse_ccsds_head_field_isolation(name: str) -> None: assert _parse_ccsds_head(_make_ccsds_head(fields)) == fields -def test_parse_ccsds_head_ignores_extra_bytes() -> None: +def test_parse_ccsds_head_warns_about_extra_bytes() -> None: """ - Test that only the first 10 bytes are decoded. + Test that extra bytes give a warning and are not decoded. - Callers do not always pass exactly 10 bytes: ``read_file_sc`` passes - the contents of the whole file to read the first header. The bytes - after the header must not change the result. + Callers are expected to pass exactly 10 bytes, so a longer input + probably means a slicing mistake in the caller, such as passing the + rest of the file. The warning makes such a mistake visible, and the + bytes after the header must still not change the result. """ header = _make_ccsds_head(_typical_fields) - assert _parse_ccsds_head(header + b"\xff" * 20) == _typical_fields + with pytest.warns(UserWarning, match="only the first 10 bytes"): + result = _parse_ccsds_head(header + b"\xff" * 20) + assert result == _typical_fields @pytest.mark.parametrize("length", range(10))