Sending UDS message (FF+CF)

.codecov.yml

@@ -3,7 +3,7 @@ codecov:
     uploads: false
 
 coverage:
-  range: 80...100
+  range: 50...100
   round: down
   precision: 2
   status:
@@ -23,6 +23,7 @@ coverage:
         flags:
           - integration-tests
       integration-tests-branch:
+        target: 50%
         flags:
           - integration-tests-branch
       performance-tests:
@@ -33,3 +34,4 @@ comment:
   layout: "reach, diff, flags, files"
   behavior: default
   require_changes: false
+  show_carryforward_flags: true

examples/can/python-can/kvaser/config.wcc

@@ -86,7 +86,7 @@
         <DIR Name="Fmt0">
           <STR Name="Name">StandardText</STR>
           <INT Name="Octal">0</INT>
-          <INT Name="Hex">0</INT>
+          <INT Name="Hex">1</INT>
           <INT Name="DeltaTime">0</INT>
           <INT Name="ShowAscii">0</INT>
           <INT Name="DefaultBase">0</INT>
@@ -162,7 +162,7 @@
         <STR Name="FileName"></STR>
         <INT Name="AutoStart">1</INT>
       </DIR>
-      <STR Name="ReverseZOrder">1,0,2</STR>
+      <STR Name="ReverseZOrder">0,1,2</STR>
     </DIR>
     <DIR Name="Connections">
       <STR Name="C0">0,1,0,0</STR>
@@ -215,7 +215,7 @@
     <INT Name="MdiMode">0</INT>
   </DIR>
   <DIR Name="Global">
-    <INT Name="UseHex">0</INT>
+    <INT Name="UseHex">1</INT>
   </DIR>
 </DIR>
 </CK>

examples/can/python-can/kvaser/python_can_timing_issue.py

@@ -14,17 +14,19 @@
 
     message = Message(data=[0x12, 0x34, 0x56, 0x78, 0x9A, 0xBC, 0xDE, 0xF0], arbitration_id=0x100)
 
-    for _ in range(100):
+    for _ in range(10):
+        timestamp_before_send = time()
         Timer(interval=0.1, function=kvaser_interface_1.send, args=(message,)).start()
 
         sent_message = buffered_reader.get_message(timeout=1)
         timestamp_after_send = time()
 
         print(f"-----------------------------------------------\n"
               f"Result:\n"
+              f"Timestamp before send: {timestamp_before_send}\n"
               f"Message timestamp: {sent_message.timestamp}\n"
               f"Current timestamp: {timestamp_after_send}\n"
-              f"Message timestamp <= Current timestamp: {sent_message.timestamp <= timestamp_after_send} (expected `True`)")
+              f"Timestamp before send <= Message timestamp <= Current timestamp: {timestamp_before_send <= sent_message.timestamp <= timestamp_after_send} (expected `True`)")
 
     kvaser_interface_1.shutdown()
     kvaser_interface_2.shutdown()

examples/can/python-can/kvaser/python_can_timing_issue_2.py

@@ -0,0 +1,32 @@
+"""UDS Issue: https://github.com/mdabrowski1990/uds/issues/228"""
+
+from time import time
+
+from can import BufferedReader, Bus, Message, Notifier
+
+if __name__ == "__main__":
+    kvaser_interface_1 = Bus(interface="kvaser", channel=0, fd=True, receive_own_messages=True)
+    kvaser_interface_2 = Bus(interface="kvaser", channel=1, fd=True, receive_own_messages=True)  # connected with bus 1
+
+    buffered_reader = BufferedReader()
+    notifier = Notifier(bus=kvaser_interface_1, listeners=[buffered_reader])
+
+    message = Message(data=[0x12, 0x34, 0x56, 0x78, 0x9A, 0xBC, 0xDE, 0xF0], arbitration_id=0x100)
+
+    for _ in range(10):
+        timestamp_before_send = time()
+        kvaser_interface_1.send(message)
+        sent_message = buffered_reader.get_message(timeout=1)
+        timestamp_after_send = time()
+
+        print(f"-----------------------------------------------\n"
+              f"Result:\n"
+              f"Timestamp before send: {timestamp_before_send}\n"
+              f"Message timestamp: {sent_message.timestamp}\n"
+              f"Current timestamp: {timestamp_after_send}\n"
+              f"Message timestamp - Timestamp before send: {sent_message.timestamp - timestamp_before_send} (expected > 0)\n"
+              f"Current timestamp - Message timestamp: {timestamp_after_send - sent_message.timestamp} (expected > 0)\n"
+              f"Timestamp before send <= Message timestamp <= Current timestamp: {timestamp_before_send <= sent_message.timestamp <= timestamp_after_send} (expected `True`)")
+
+    kvaser_interface_1.shutdown()
+    kvaser_interface_2.shutdown()

uds/segmentation/can_segmenter.py

@@ -12,6 +12,7 @@
     CanConsecutiveFrameHandler,
     CanDlcHandler,
     CanFirstFrameHandler,
+    CanFlowStatus,
     CanSingleFrameHandler,
     PacketAIParamsAlias,
 )
@@ -161,6 +162,78 @@ def filler_byte(self, value: int):
         validate_raw_byte(value)
         self.__filler_byte: int = value
 
+    def __physical_segmentation(self, message: UdsMessage) -> Tuple[CanPacket, ...]:
+        """
+        Segment physically addressed diagnostic message.
+
+        :param message: UDS message to divide into UDS packets.
+
+        :raise SegmentationError: Provided diagnostic message cannot be segmented.
+
+        :return: CAN packets that are an outcome of UDS message segmentation.
+        """
+        message_payload_size = len(message.payload)
+        if message_payload_size > CanFirstFrameHandler.MAX_LONG_FF_DL_VALUE:
+            raise SegmentationError("Provided diagnostic message cannot be segmented to CAN Packet as it is too big "
+                                    "to transmit it over CAN bus.")
+        if message_payload_size <= CanSingleFrameHandler.get_max_payload_size(addressing_format=self.addressing_format,
+                                                                              dlc=self.dlc):
+            single_frame = CanPacket(packet_type=CanPacketType.SINGLE_FRAME,
+                                     payload=message.payload,
+                                     filler_byte=self.filler_byte,
+                                     dlc=None if self.use_data_optimization else self.dlc,
+                                     **self.tx_packets_physical_ai)
+            return (single_frame,)
+        ff_payload_size = CanFirstFrameHandler.get_payload_size(
+            addressing_format=self.addressing_format,
+            dlc=self.dlc,
+            long_ff_dl_format=message_payload_size > CanFirstFrameHandler.MAX_SHORT_FF_DL_VALUE)
+        first_frame = CanPacket(packet_type=CanPacketType.FIRST_FRAME,
+                                payload=message.payload[:ff_payload_size],
+                                dlc=self.dlc,
+                                data_length=message_payload_size,
+                                **self.tx_packets_physical_ai)
+        cf_payload_size = CanConsecutiveFrameHandler.get_max_payload_size(addressing_format=self.addressing_format,
+                                                                          dlc=self.dlc)
+        total_cfs_number = (message_payload_size - ff_payload_size + cf_payload_size - 1) // cf_payload_size
+        consecutive_frames = []
+        for cf_index in range(total_cfs_number):
+            sequence_number = (cf_index + 1) % 0x10
+            payload_i_start = ff_payload_size + cf_index * cf_payload_size
+            payload_i_stop = payload_i_start + cf_payload_size
+            consecutive_frame = CanPacket(packet_type=CanPacketType.CONSECUTIVE_FRAME,
+                                          payload=message.payload[payload_i_start: payload_i_stop],
+                                          dlc=None if self.use_data_optimization and cf_index == total_cfs_number - 1
+                                          else self.dlc,
+                                          sequence_number=sequence_number,
+                                          filler_byte=self.filler_byte,
+                                          **self.tx_packets_physical_ai)
+            consecutive_frames.append(consecutive_frame)
+        return (first_frame, *consecutive_frames)
+
+    def __functional_segmentation(self, message: UdsMessage) -> Tuple[CanPacket, ...]:
+        """
+        Segment functionally addressed diagnostic message.
+
+        :param message: UDS message to divide into UDS packets.
+
+        :raise SegmentationError: Provided diagnostic message cannot be segmented.
+
+        :return: CAN packets that are an outcome of UDS message segmentation.
+        """
+        max_payload_size = CanSingleFrameHandler.get_max_payload_size(addressing_format=self.addressing_format,
+                                                                      dlc=self.dlc)
+        message_payload_size = len(message.payload)
+        if message_payload_size > max_payload_size:
+            raise SegmentationError("Provided diagnostic message cannot be segmented using functional addressing "
+                                    "as it will not fit into a Single Frame.")
+        single_frame = CanPacket(packet_type=CanPacketType.SINGLE_FRAME,
+                                 payload=message.payload,
+                                 filler_byte=self.filler_byte,
+                                 dlc=None if self.use_data_optimization else self.dlc,
+                                 **self.tx_packets_functional_ai)
+        return (single_frame,)
+
     # TODO: consider moving this method to AbstractAddressingInformation (if defined)
     def is_input_packet(self, can_id: int, data: RawBytesAlias) -> Optional[AddressingType]:  # type: ignore # noqa
         """
@@ -225,6 +298,29 @@ def is_desegmented_message(self, packets: PacketsContainersSequence) -> bool:
             return payload_bytes_found >= total_payload_size  # type: ignore
         raise NotImplementedError(f"Unknown packet type received: {packets[0].packet_type}")
 
+    def get_flow_control_packet(self,
+                                flow_status: CanFlowStatus,
+                                block_size: Optional[int] = None,
+                                st_min: Optional[int] = None) -> CanPacket:
+        """
+        Create Flow Control CAN packet.
+
+        :param flow_status: Value of Flow Status parameter.
+        :param block_size: Value of Block Size parameter.
+            This parameter is only required with ContinueToSend Flow Status, leave None otherwise.
+        :param st_min: Value of Separation Time minimum (STmin) parameter.
+            This parameter is only required with ContinueToSend Flow Status, leave None otherwise.
+
+        :return: Flow Control CAN packet with provided parameters.
+        """
+        return CanPacket(packet_type=CanPacketType.FLOW_CONTROL,
+                         flow_status=flow_status,
+                         block_size=block_size,
+                         st_min=st_min,
+                         filler_byte=self.filler_byte,
+                         dlc=None if self.use_data_optimization else self.dlc,
+                         **self.tx_packets_physical_ai)
+
     def desegmentation(self, packets: PacketsContainersSequence) -> Union[UdsMessage, UdsMessageRecord]:
         """
         Perform desegmentation of CAN packets.
@@ -276,75 +372,3 @@ def segmentation(self, message: UdsMessage) -> Tuple[CanPacket, ...]:
         if message.addressing_type == AddressingType.FUNCTIONAL:
             return self.__functional_segmentation(message)
         raise NotImplementedError(f"Unknown addressing type received: {message.addressing_type}")
-
-    def __physical_segmentation(self, message: UdsMessage) -> Tuple[CanPacket, ...]:
-        """
-        Segment physically addressed diagnostic message.
-
-        :param message: UDS message to divide into UDS packets.
-
-        :raise SegmentationError: Provided diagnostic message cannot be segmented.
-
-        :return: CAN packets that are an outcome of UDS message segmentation.
-        """
-        message_payload_size = len(message.payload)
-        if message_payload_size > CanFirstFrameHandler.MAX_LONG_FF_DL_VALUE:
-            raise SegmentationError("Provided diagnostic message cannot be segmented to CAN Packet as it is too big "
-                                    "to transmit it over CAN bus.")
-        if message_payload_size <= CanSingleFrameHandler.get_max_payload_size(addressing_format=self.addressing_format,
-                                                                              dlc=self.dlc):
-            single_frame = CanPacket(packet_type=CanPacketType.SINGLE_FRAME,
-                                     payload=message.payload,
-                                     filler_byte=self.filler_byte,
-                                     dlc=None if self.use_data_optimization else self.dlc,
-                                     **self.tx_packets_physical_ai)
-            return (single_frame,)
-        ff_payload_size = CanFirstFrameHandler.get_payload_size(
-            addressing_format=self.addressing_format,
-            dlc=self.dlc,
-            long_ff_dl_format=message_payload_size > CanFirstFrameHandler.MAX_SHORT_FF_DL_VALUE)
-        first_frame = CanPacket(packet_type=CanPacketType.FIRST_FRAME,
-                                payload=message.payload[:ff_payload_size],
-                                dlc=self.dlc,
-                                data_length=message_payload_size,
-                                **self.tx_packets_physical_ai)
-        cf_payload_size = CanConsecutiveFrameHandler.get_max_payload_size(addressing_format=self.addressing_format,
-                                                                          dlc=self.dlc)
-        total_cfs_number = (message_payload_size - ff_payload_size + cf_payload_size - 1) // cf_payload_size
-        consecutive_frames = []
-        for cf_index in range(total_cfs_number):
-            sequence_number = (cf_index + 1) % 0x10
-            payload_i_start = ff_payload_size + cf_index * cf_payload_size
-            payload_i_stop = payload_i_start + cf_payload_size
-            consecutive_frame = CanPacket(packet_type=CanPacketType.CONSECUTIVE_FRAME,
-                                          payload=message.payload[payload_i_start: payload_i_stop],
-                                          dlc=None if self.use_data_optimization and cf_index == total_cfs_number - 1
-                                          else self.dlc,
-                                          sequence_number=sequence_number,
-                                          filler_byte=self.filler_byte,
-                                          **self.tx_packets_physical_ai)
-            consecutive_frames.append(consecutive_frame)
-        return (first_frame, *consecutive_frames)
-
-    def __functional_segmentation(self, message: UdsMessage) -> Tuple[CanPacket, ...]:
-        """
-        Segment functionally addressed diagnostic message.
-
-        :param message: UDS message to divide into UDS packets.
-
-        :raise SegmentationError: Provided diagnostic message cannot be segmented.
-
-        :return: CAN packets that are an outcome of UDS message segmentation.
-        """
-        max_payload_size = CanSingleFrameHandler.get_max_payload_size(addressing_format=self.addressing_format,
-                                                                      dlc=self.dlc)
-        message_payload_size = len(message.payload)
-        if message_payload_size > max_payload_size:
-            raise SegmentationError("Provided diagnostic message cannot be segmented using functional addressing "
-                                    "as it will not fit into a Single Frame.")
-        single_frame = CanPacket(packet_type=CanPacketType.SINGLE_FRAME,
-                                 payload=message.payload,
-                                 filler_byte=self.filler_byte,
-                                 dlc=None if self.use_data_optimization else self.dlc,
-                                 **self.tx_packets_functional_ai)
-        return (single_frame,)