diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..560a9d9 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,78 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project + +`can-j1939` is a Python implementation of the SAE J1939 protocol stack on top of +[python-can](https://python-can.readthedocs.org/). It supports both J1939-21 and J1939-22 (J1939-FD) +data link layers, including transport protocols (BAM, CMDT / RTS-CTS), address claiming, and a +number of diagnostic messages (DM1, DM11, DM14, DM22). + +## Common commands + +```bash +# Install the package (editable for development) +pip install -e . + +# Run the full test suite (matches CI) +pytest . --pyargs + +# Run a single test file / test +pytest test/test_ecu.py +pytest test/test_memory_access.py::TestMemoryAccess::test_some_name -v +``` + +CI runs `pytest . --pyargs` on Python 3.10 across Ubuntu/macOS/Windows +(`.github/workflows/CI.yml`). + +## Architecture + +The stack is layered: an **ECU** owns a **data-link layer** object and one or more +**ControllerApplications**. Background work runs on a dedicated job thread. + +- `j1939/electronic_control_unit.py` — `ElectronicControlUnit` is the entry point. It owns the + `can.Bus`, a `MessageListener`, a job thread (`_async_job_thread`) that drives timers and + transport-protocol timeouts, and a list of subscribers. The `data_link_layer` constructor arg + (`'j1939-21'` or `'j1939-22'`) selects which DLL is instantiated. The ECU passes the DLL a + small surface of callbacks: `send_message`, `_job_thread_wakeup`, `_notify_subscribers`, + `_is_message_acceptable`. For tests, `send_message=` can be injected to bypass real CAN I/O. +- `j1939/j1939_21.py` and `j1939/j1939_22.py` — the two DLL implementations. They share the + callback signature above and implement the transport protocols (TP-BAM, TP-CMDT / RTS-CTS, + and for J1939-22 the FD multi-session variants and Multi-PG / FEFF). Changes that touch + protocol behaviour usually need parallel updates in both files. +- `j1939/controller_application.py` — `ControllerApplication` (CA) implements J1939/81 address + claiming, state machine (`NONE` → `WAITING_VETO` → `NORMAL` / `CANNOT_CLAIM`), per-CA + subscriptions, and `send_pgn` (which dispatches to the ECU's DLL). +- `j1939/name.py`, `j1939/parameter_group_number.py`, `j1939/message_id.py` — value objects for + the J1939 NAME, PGN encoding, and 29-bit CAN identifier framing. +- `j1939/diagnostic_messages.py`, `j1939/memory_access.py`, `j1939/Dm14Query.py`, + `j1939/Dm14Server.py`, `j1939/error_info.py` — diagnostic-message support (DM1/DM11/DM14/DM22), + including the DM14 memory-access client (`Dm14Query`) and server (`Dm14Server`). +- `j1939/__init__.py` is the public API surface — anything users are expected to import lives + here. + +### Threading model + +All I/O and protocol timing flows through the ECU's job thread. The DLL never blocks on I/O +itself — it enqueues work and calls `_job_thread_wakeup` to nudge the thread. Callbacks +registered via `ca.subscribe(...)` or `ca.add_timer(...)` run on that job thread, so they +must not block. + +### Tests + +- `test/` holds unit tests. `test_helpers/feeder.py` provides the `Feeder` fixture (registered + in `test_helpers/conftest.py`) which is the standard way to drive the stack from tests: it + replaces `ElectronicControlUnit.send_message` with a simulated bus, lets the test queue + expected RX/TX messages and PDUs in order, and asserts that the stack produces the expected + TX sequence. New protocol-level tests should follow that pattern instead of mocking + `python-can` directly. +- `test_helpers/feeder.AcceptAllCA` is a CA subclass with `message_acceptable` overridden to + accept everything — use it when a test needs to receive peer-to-peer messages without setting + up a real claim. + +### Examples + +`examples/` contains runnable scripts mirroring the README quick-start (simple receive, own CA +producer, transport protocols, multi-PG, diagnostic messages). When adding a new public +feature, prefer extending an existing example over inventing a new pattern in the docs. diff --git a/README.rst b/README.rst index 478b555..f8a0118 100644 --- a/README.rst +++ b/README.rst @@ -72,7 +72,7 @@ Features * correct timeout and deadline handling * (under construction) almost complete testcoverage * diagnostic messages (see https://github.com/juergenH87/python-can-j1939/tree/master/examples/diagnostic_message.py) - - support of DM1 Tool and ECU functionaliy + - support of DM1 Tool and ECU functionaliy (all four SAE J1939-73 SPN conversion methods: 1, 2, 3, 4) - support of DM11 Tool functionaliy - support of DM22 Tool functionaliy diff --git a/examples/diagnostic_message.py b/examples/diagnostic_message.py index b76fa34..3008053 100644 --- a/examples/diagnostic_message.py +++ b/examples/diagnostic_message.py @@ -54,7 +54,7 @@ def dm1_before_send(): :return: list of dictionaries of all DTCs included in DM1 - :rtype: list of dic: 'spn', 'fmi', 'oc' + :rtype: list of dic: 'spn', 'fmi', 'oc', 'cm' """ lamp_status = {} # get lamp status (optional, if status not enter, lamp is switched off) @@ -65,9 +65,12 @@ def dm1_before_send(): # add all active DTCs # if no DTC is active return empty list + # 'cm' is the SAE J1939-73 SPN conversion method (1, 2, 3, or 4). + # Defaults to 4 (current standard) when omitted. dtc_list = [] - dtc_list.append({'spn': 123, 'fmi': 31}) # occurrence counter is set to 0 - dtc_list.append({'spn': 456, 'fmi': 1, 'oc': 132}) # with optional occurrence counter + dtc_list.append({'spn': 123, 'fmi': 31}) # CM defaults to 4 + dtc_list.append({'spn': 456, 'fmi': 1, 'oc': 132}) # with occurrence counter + dtc_list.append({'spn': 789, 'fmi': 2, 'oc': 5, 'cm': 3}) # legacy CM 3 layout return lamp_status, dtc_list diff --git a/j1939/diagnostic_messages.py b/j1939/diagnostic_messages.py index 9d6ac6a..1cfb3e3 100644 --- a/j1939/diagnostic_messages.py +++ b/j1939/diagnostic_messages.py @@ -5,23 +5,66 @@ class DTC: """ - Parser for J1939 DTC (Diagnostic Trouble Code) + Parser/encoder for J1939 DTC (Diagnostic Trouble Code). + + Supports the four SAE J1939-73 SPN conversion methods: + - CM 1: SPN MSBs in byte 1, mid in byte 2, LSBs+FMI in byte 3, CM bit = 1 + - CM 2: SPN mid in byte 1, MSBs in byte 2, LSBs+FMI in byte 3, CM bit = 1 + - CM 3: SPN LSBs/mid/MSBs in bytes 1/2/3 (modern layout), CM bit = 1 + - CM 4: same byte layout as CM 3, CM bit = 0 (current standard) + + The on-wire CM bit only distinguishes {1,2,3} (bit=1) from {4} (bit=0). + CM 1 vs CM 2 vs CM 3 are not separable from the bytes alone; when + decoding raw bytes with CM bit = 1, the caller must indicate which one + was used (defaults to CM 3 — the most common legacy layout). """ - def __init__(self, dtc=None, spn=None, fmi=None, oc=0): - if dtc != None: + def __init__(self, dtc=None, spn=None, fmi=None, oc=0, cm=4): + if dtc is not None: + self._cm = cm self._dtc = dtc - self._spn = ((dtc & 0xFFFF) | ((dtc >> 5) & 0x70000)) - self._fmi = ((dtc >> 16) & 0x1F) - self._oc = ((dtc >> 24) & 0x7f) - self._cm = ((dtc >> 31) & 0x01) - if self._cm != 0: - logger.error("DM01: deprecated spn conversion modes are not supported") + self._oc = ((dtc >> 24) & 0x7F) + cm_bit = ((dtc >> 31) & 0x01) + b1 = dtc & 0xFF + b2 = (dtc >> 8) & 0xFF + b3 = (dtc >> 16) & 0xFF + self._fmi = b3 & 0x1F + spn_low3 = (b3 >> 5) & 0x07 + if cm in (3, 4): + self._spn = b1 | (b2 << 8) | (spn_low3 << 16) + elif cm == 1: + # b1 = SPN[18:11], b2 = SPN[10:3], b3[7:5] = SPN[2:0] + self._spn = (b1 << 11) | (b2 << 3) | spn_low3 + elif cm == 2: + # b1 = SPN[10:3], b2 = SPN[18:11], b3[7:5] = SPN[2:0] + self._spn = (b2 << 11) | (b1 << 3) | spn_low3 + else: + raise ValueError(f"Invalid conversion method: {cm}. Must be 1, 2, 3, or 4.") + # Sanity-check the CM bit against the requested method + expected_cm_bit = 0 if cm == 4 else 1 + if cm_bit != expected_cm_bit: + logger.warning("DM01: CM bit %d does not match requested conversion method %d", cm_bit, cm) else: - self._dtc = ((spn & 0xFFFF) | ((spn & 0x70000) << 5) | ((fmi & 0x1F) << 16) | ((oc & 0x7F) << 24)) + if cm not in (1, 2, 3, 4): + raise ValueError(f"Invalid conversion method: {cm}. Must be 1, 2, 3, or 4.") self._spn = spn self._fmi = fmi self._oc = oc - self._cm = 0 + self._cm = cm + if cm == 1: + b1 = (spn >> 11) & 0xFF + b2 = (spn >> 3) & 0xFF + elif cm == 2: + b1 = (spn >> 3) & 0xFF + b2 = (spn >> 11) & 0xFF + else: # cm in (3, 4) + b1 = spn & 0xFF + b2 = (spn >> 8) & 0xFF + b3 = (((spn >> 16) & 0x07) << 5) | (fmi & 0x1F) if cm in (3, 4) \ + else ((spn & 0x07) << 5) | (fmi & 0x1F) + b4 = oc & 0x7F + if cm != 4: + b4 |= 0x80 + self._dtc = b1 | (b2 << 8) | (b3 << 16) | (b4 << 24) @property def spn(self): @@ -57,7 +100,7 @@ def oc(self): def cm(self): """ :return: - SPN conversion mode + SPN conversion method (1, 2, 3, or 4 per SAE J1939-73) :rtype: int """ @@ -127,16 +170,24 @@ class Dm1: """ _msg_subscriber_added = False - def __init__(self, ca: j1939.ControllerApplication): + def __init__(self, ca: j1939.ControllerApplication, rx_cm_bit_set: int = 3): """ :param obj ca: j1939 controller application + :param int rx_cm_bit_set: + SPN conversion method (1, 2, or 3) to assume when a received DTC + has its CM bit set. The on-wire CM bit cannot distinguish CMs 1, + 2 and 3 — only between {1,2,3} (bit=1) and 4 (bit=0). Defaults to + 3 (the most common legacy layout). CM 4 is auto-detected. """ + if rx_cm_bit_set not in (1, 2, 3): + raise ValueError(f"rx_cm_bit_set must be 1, 2, or 3 (got {rx_cm_bit_set})") self._pgn = j1939.ParameterGroupNumber.PGN.DM01 self._lamp_status = {} self._dtc_dic_list = [] self._data = [] self._subscribers = [] self._ca = ca + self._rx_cm_bit_set = rx_cm_bit_set def subscribe(self, callback): """Add the given callback to the Dm1 message notification stream. @@ -230,8 +281,9 @@ def _send(self, cookie): # optional arguments if dtc_dic.get('oc') == None: dtc_dic['oc'] = 0 + cm = dtc_dic.get('cm', 4) - dtc = DTC(spn=dtc_dic['spn'], fmi=dtc_dic['fmi'], oc=dtc_dic['oc']).dtc + dtc = DTC(spn=dtc_dic['spn'], fmi=dtc_dic['fmi'], oc=dtc_dic['oc'], cm=cm).dtc self._data.append(dtc & 0xFF) self._data.append((dtc >> 8) & 0xFF) self._data.append((dtc >> 16) & 0xFF) @@ -290,8 +342,9 @@ def _parse_dm1_receive_data(self): # so we should not add this to the dtc list since it is not a valid dtc continue - dtc = DTC(dtc=dtc_int) - self._dtc_dic_list.append( {'spn': dtc.spn, 'fmi': dtc.fmi, 'oc': dtc.oc } ) + cm = 4 if ((dtc_int >> 31) & 0x01) == 0 else self._rx_cm_bit_set + dtc = DTC(dtc=dtc_int, cm=cm) + self._dtc_dic_list.append( {'spn': dtc.spn, 'fmi': dtc.fmi, 'oc': dtc.oc, 'cm': dtc.cm } ) def _notify_subscribers(self, sa, timestamp): for callback in self._subscribers: diff --git a/j1939/j1939_22.py b/j1939/j1939_22.py index c55cf93..a314273 100644 --- a/j1939/j1939_22.py +++ b/j1939/j1939_22.py @@ -2,7 +2,6 @@ from .message_id import MessageId, FrameFormat import logging import time -import numpy as np logger = logging.getLogger(__name__) @@ -267,13 +266,8 @@ def send_pgn(self, data_page, pdu_format, pdu_specific, priority, src_address, d if priority == None: priority = 7 # get chunks from data - full_tp_size_packages = int(data_length/self.DataLength.TP) - arr = np.array(data) - list_of_arr = np.split(arr, [full_tp_size_packages*self.DataLength.TP]) - arr = np.reshape(list_of_arr[0], (-1,self.DataLength.TP)) - data_list = arr.tolist() - if len(list_of_arr) > 1: - data_list.append(list_of_arr[1].tolist()) + chunk_size = self.DataLength.TP + data_list = [list(data[i:i + chunk_size]) for i in range(0, data_length, chunk_size)] # if the PF is between 240 and 255, the message can only be broadcast if dest_address == ParameterGroupNumber.Address.GLOBAL: diff --git a/setup.py b/setup.py index 8885ff7..0594b8c 100644 --- a/setup.py +++ b/setup.py @@ -27,7 +27,6 @@ ], install_requires=[ "python-can >= 3.3.4", - "numpy >= 1.17.0", "pytest >= 6.2.5", ], include_package_data=True, diff --git a/test/test_dtc_conversion_methods.py b/test/test_dtc_conversion_methods.py new file mode 100644 index 0000000..f646c7a --- /dev/null +++ b/test/test_dtc_conversion_methods.py @@ -0,0 +1,85 @@ +"""Tests for the SAE J1939-73 SPN conversion methods (CM 1, 2, 3, 4).""" +import pytest + +from j1939.diagnostic_messages import DTC + + +@pytest.mark.parametrize("cm", [1, 2, 3, 4]) +@pytest.mark.parametrize("spn", [0, 123, 456, 0x12345, 0x3FFFF]) +@pytest.mark.parametrize("fmi", [0, 1, 5, 31]) +@pytest.mark.parametrize("oc", [0, 1, 42, 127]) +def test_dtc_round_trip(cm, spn, fmi, oc): + """Encoded DTC bytes must decode back to the same SPN/FMI/OC/CM.""" + encoded = DTC(spn=spn, fmi=fmi, oc=oc, cm=cm) + decoded = DTC(dtc=encoded.dtc, cm=cm) + assert decoded.spn == spn + assert decoded.fmi == fmi + assert decoded.oc == oc + assert decoded.cm == cm + + +@pytest.mark.parametrize("cm,expected_cm_bit", [(1, 1), (2, 1), (3, 1), (4, 0)]) +def test_cm_bit_on_wire(cm, expected_cm_bit): + """Only CM 4 has the CM bit cleared; CMs 1/2/3 set it.""" + d = DTC(spn=1000, fmi=5, oc=3, cm=cm) + assert ((d.dtc >> 31) & 0x01) == expected_cm_bit + + +def test_cm1_byte_layout_matches_reference(): + """CM 1 layout: b1=SPN[18:11], b2=SPN[10:3], b3=SPN[2:0]|FMI, b4=OC|CM.""" + spn, fmi, oc = 0x12345, 5, 3 + d = DTC(spn=spn, fmi=fmi, oc=oc, cm=1) + b1 = d.dtc & 0xFF + b2 = (d.dtc >> 8) & 0xFF + b3 = (d.dtc >> 16) & 0xFF + b4 = (d.dtc >> 24) & 0xFF + assert b1 == (spn >> 11) & 0xFF + assert b2 == (spn >> 3) & 0xFF + assert b3 == (((spn & 0x07) << 5) | (fmi & 0x1F)) + assert b4 == ((oc & 0x7F) | 0x80) + + +def test_cm2_byte_layout_matches_reference(): + """CM 2 layout: b1=SPN[10:3], b2=SPN[18:11], b3=SPN[2:0]|FMI, b4=OC|CM.""" + spn, fmi, oc = 0x12345, 5, 3 + d = DTC(spn=spn, fmi=fmi, oc=oc, cm=2) + b1 = d.dtc & 0xFF + b2 = (d.dtc >> 8) & 0xFF + b3 = (d.dtc >> 16) & 0xFF + b4 = (d.dtc >> 24) & 0xFF + assert b1 == (spn >> 3) & 0xFF + assert b2 == (spn >> 11) & 0xFF + assert b3 == (((spn & 0x07) << 5) | (fmi & 0x1F)) + assert b4 == ((oc & 0x7F) | 0x80) + + +@pytest.mark.parametrize("cm", [3, 4]) +def test_cm3_cm4_byte_layout(cm): + """CM 3/4 layout: SPN packed little-endian in b1/b2 with top 3 bits in b3.""" + spn, fmi, oc = 0x12345, 5, 3 + d = DTC(spn=spn, fmi=fmi, oc=oc, cm=cm) + b1 = d.dtc & 0xFF + b2 = (d.dtc >> 8) & 0xFF + b3 = (d.dtc >> 16) & 0xFF + b4 = (d.dtc >> 24) & 0xFF + assert b1 == spn & 0xFF + assert b2 == (spn >> 8) & 0xFF + assert b3 == ((((spn >> 16) & 0x07) << 5) | (fmi & 0x1F)) + expected_b4 = oc & 0x7F + if cm == 3: + expected_b4 |= 0x80 + assert b4 == expected_b4 + + +def test_invalid_cm_raises(): + with pytest.raises(ValueError): + DTC(spn=1, fmi=1, oc=0, cm=5) + with pytest.raises(ValueError): + DTC(dtc=0x12345678, cm=0) + + +def test_default_cm_is_4(): + """Backward compatibility: omitting `cm` produces the modern CM 4 layout.""" + d = DTC(spn=0x12345, fmi=5, oc=3) + assert d.cm == 4 + assert ((d.dtc >> 31) & 0x01) == 0 diff --git a/test/test_j1939_22.py b/test/test_j1939_22.py new file mode 100644 index 0000000..254f1ef --- /dev/null +++ b/test/test_j1939_22.py @@ -0,0 +1,173 @@ +""" +Tests for J1939-22 transport protocol chunking logic. + +This module tests the data chunking algorithm used in J1939-22 for splitting +large messages into transport protocol segments of 60 bytes each. +""" +import pytest + +from j1939.j1939_22 import J1939_22 +from j1939.message_id import FrameFormat +from test_helpers.conftest import feeder + + +class TestChunkingAlgorithm: + """Isolated tests for the data chunking algorithm.""" + + @staticmethod + def chunk_data(data, chunk_size): + """Pure-Python chunking implementation matching j1939_22.py:send_pgn()""" + data_length = len(data) + return [list(data[i:i + chunk_size]) for i in range(0, data_length, chunk_size)] + + @pytest.mark.parametrize("data_length,expected_chunks,expected_last_chunk_size", [ + (60, 1, 60), + (61, 2, 1), + (119, 2, 59), + (120, 2, 60), + (121, 3, 1), + (180, 3, 60), + (181, 4, 1), + ]) + def test_chunk_sizes(self, data_length, expected_chunks, expected_last_chunk_size): + """Verify correct chunk count and sizes for various data lengths.""" + data = list(range(data_length)) + result = self.chunk_data(data, J1939_22.DataLength.TP) + + assert len(result) == expected_chunks + assert len(result[-1]) == expected_last_chunk_size + + def test_data_integrity(self): + """All original bytes are present after chunking, in correct order.""" + data = list(range(2560)) # Large data set: 43 chunks + result = self.chunk_data(data, J1939_22.DataLength.TP) + + # Verify chunk count + expected_chunks = 2560 // J1939_22.DataLength.TP + (1 if 2560 % J1939_22.DataLength.TP else 0) + assert len(result) == expected_chunks + + # Verify all data preserved in order + reconstructed = [] + for chunk in result: + reconstructed.extend(chunk) + assert reconstructed == data + + def test_chunk_count_matches_num_segments_formula(self): + """Verify chunking matches the num_segments formula used in j1939_22.py.""" + for data_length in [60, 61, 119, 120, 121, 180, 500, 1000]: + data = [i % 256 for i in range(data_length)] + + result = self.chunk_data(data, J1939_22.DataLength.TP) + + # Formula from j1939_22.py + expected = int(data_length / J1939_22.DataLength.TP) + ((data_length % J1939_22.DataLength.TP) != 0) + + assert len(result) == expected, f"data_length={data_length}" + + +class TestJ1939_22Integration: + """Integration tests for J1939-22 chunking through send_pgn.""" + + @staticmethod + def create_j1939_22(): + """Create a J1939_22 instance with mock callbacks.""" + return J1939_22( + send_message=lambda *args, **kwargs: None, + job_thread_wakeup=lambda: None, + notify_subscribers=lambda *args: None, + max_cmdt_packets=16, + minimum_tp_rts_cts_dt_interval=None, + minimum_tp_bam_dt_interval=0.010, + ecu_is_message_acceptable=lambda dest: True + ) + + def test_short_message_not_chunked(self, feeder): + """Data <= J1939_22.DataLength.TP bytes uses multi-pg path, not TP chunking.""" + feeder.accept_all_messages() + j1939_22 = self.create_j1939_22() + + result = j1939_22.send_pgn( + data_page=0, pdu_format=0xFE, pdu_specific=0xFF, + priority=7, src_address=0x01, data=list(range(J1939_22.DataLength.TP)), + time_limit=0, frame_format=FrameFormat.CEFF + ) + + assert result is True + assert len(j1939_22._snd_buffer) == 0 + + def test_bam_broadcast_chunking(self, feeder): + """BAM broadcast correctly chunks data and verifies integrity.""" + feeder.accept_all_messages() + j1939_22 = self.create_j1939_22() + + test_data = list(range(121)) # 3 chunks: 60 + 60 + 1 + + result = j1939_22.send_pgn( + data_page=0, pdu_format=0xFE, pdu_specific=0xFF, + priority=7, src_address=0x01, data=test_data, + time_limit=0, frame_format=FrameFormat.CEFF + ) + + assert result is True + buffer = list(j1939_22._snd_buffer.values())[0] + + assert buffer['num_segments'] == 3 + assert len(buffer['data']) == 3 + assert [len(chunk) for chunk in buffer['data']] == [60, 60, 1] + + # Verify data integrity + reconstructed = [] + for chunk in buffer['data']: + reconstructed.extend(chunk) + assert reconstructed == test_data + + def test_rts_cts_peer_to_peer_chunking(self, feeder): + """RTS/CTS peer-to-peer uses different code path but chunks correctly.""" + feeder.accept_all_messages() + j1939_22 = self.create_j1939_22() + + test_data = list(range(180)) # 3 chunks of 60 each + + result = j1939_22.send_pgn( + data_page=0, pdu_format=0xDF, pdu_specific=0x04, # PDU1 = peer-to-peer + priority=7, src_address=0x01, data=test_data, + time_limit=0, frame_format=1 + ) + + assert result is True + buffer = list(j1939_22._snd_buffer.values())[0] + + assert buffer['num_segments'] == 3 + assert all(len(chunk) == 60 for chunk in buffer['data']) + + reconstructed = [] + for chunk in buffer['data']: + reconstructed.extend(chunk) + assert reconstructed == test_data + + @pytest.mark.parametrize("data_length,expected_segments", [ + (61, 2), + (120, 2), + (121, 3), + (180, 3), + (240, 4), + (500, 9), + ]) + def test_various_data_sizes(self, feeder, data_length, expected_segments): + """Parametrized test for segment count across various data sizes.""" + feeder.accept_all_messages() + j1939_22 = self.create_j1939_22() + + test_data = [i % 256 for i in range(data_length)] + + result = j1939_22.send_pgn( + data_page=0, pdu_format=0xFE, pdu_specific=0xFF, + priority=7, src_address=0x01, data=test_data, + time_limit=0, frame_format=FrameFormat.CEFF + ) + + assert result is True + buffer = list(j1939_22._snd_buffer.values())[0] + + assert buffer['num_segments'] == expected_segments + assert len(buffer['data']) == expected_segments