add support for container-pdus without sub-pdu Header (=static container-pdu). signals of such sub-pdus now get the correct start_bit

Author: Markus Konrad

src/canmatrix/canmatrix.py

@@ -833,6 +833,8 @@ class Pdu(object):
     Whereas a PDU is the same than a frame on CAN bus, at flexray a frame may consist of
     multiple PDUs (a bit like multiple signal layout for multiplexed can frames).
     This class is only used for flexray busses.
+    Note: since container-pdus are supported for arxml, this class is also used for arxml (but only
+          for sub-pdus of container-pdus).
     """
 
     name = attr.ib(default="")  # type: str
@@ -844,6 +846,8 @@ class Pdu(object):
     signals = attr.ib(factory=list)  # type: typing.MutableSequence[Signal]
     signalGroups = attr.ib(factory=list)  # type: typing.MutableSequence[SignalGroup]
     cycle_time = attr.ib(default=0)  # type: int
+    # offset is used for arxml, sub-pdu inside a static-container-pdu
+    offset_bytes = attr.ib(default=0)  # type: int
 
     def add_signal(self, signal):
         # type: (Signal) -> Signal
@@ -1491,20 +1495,35 @@ def unpack(self, data: bytes,
                     f"Received message 0x{msg_id:04X} with wrong data size: {rx_length} instead of {self.size}")
 
         if self.is_pdu_container:
+            # note: PDU-Container without header is possible for ARXML-Container-PDUs with NO-HEADER
+            #       that mean this are not dynamic Container-PDUs rather than static ones. (each sub-pdu has
+            #       a fixed offset in the container)
+            header_signals = []
             header_id_signal = self.signal_by_name("Header_ID")
             header_dlc_signal = self.signal_by_name("Header_DLC")
-            if header_id_signal is None or header_dlc_signal is None:
-                raise DecodingConatainerPdu(
-                    'Received message 0x{:08X} without Header_ID or '
-                    'Header_DLC signal'.format(self.arbitration_id.id)
-                )
+
+            if header_id_signal is not None:
+                header_signals.append(header_id_signal)
+                _header_id_signal_size = header_id_signal.size
+            else:
+                _header_id_signal_size = 0
+            if header_dlc_signal is not None:
+                header_signals.append(header_dlc_signal)
+                _header_dlc_signal_size = header_dlc_signal.size
+            else:
+                _header_dlc_signal_size = 0
             # TODO: may be we need to check that ID/DLC signals are contiguous
-            header_size = header_id_signal.size + header_dlc_signal.size
+            if len(header_signals) > 0 and len(header_signals) != 2:
+                raise DecodingContainerPdu(
+                        'Received message 0x{:08X} with incorrect Header-Defintiion. '
+                        'Header_ID signal or Header_DLC is missing'.format(self.arbitration_id.id)
+                    )
+            header_size = _header_id_signal_size + _header_dlc_signal_size
             little, big = self.bytes_to_bitstrings(data)
             size = self.size * 8
             return_dict = dict({"pdus": []})
             # decode signal which are not in PDUs
-            signals = [s for s in self.signals if s not in [header_id_signal, header_dlc_signal]]
+            signals = [s for s in self.signals if s not in header_signals]
             if signals:
                 unpacked = self.bitstring_to_signal_list(signals, big, little, size)
                 for s, v in zip(signals, unpacked):
@@ -1527,20 +1546,52 @@ def unpack(self, data: bytes,
                         return_dict[s.name] = []
                     return_dict[s.name].append(DecodedSignal(v, s))
                 pdu = self.pdu_by_id(pdu_id)
+            offset = header_id_signal.start_bit if header_id_signal is not None else 0
+            no_header_next_pdu_idx = 0
+            # decode as long as there is data left to decode (if there is a header), or as long as there are sub-pdus
+            # left to decode (in case of static-container without pdu-headers)
+            while (offset + header_size) < size and no_header_next_pdu_idx < len(self.pdus):
+                if len(header_signals) > 0:
+                    unpacked = self.bitstring_to_signal_list(
+                        header_signals,
+                        big[offset:offset + header_size],
+                        little[size - offset - header_size:size - offset],
+                        header_size
+                    )
+                    offset += header_size
+                    pdu_id = unpacked[0]
+                    pdu_dlc = unpacked[1]
+                    for s, v in zip(header_signals, unpacked):
+                        if s.name not in return_dict:
+                            return_dict[s.name] = []
+                        return_dict[s.name].append(DecodedSignal(v, s))
+                    pdu = self.pdu_by_id(pdu_id)
+                else:
+                    # if there is no pdu-header, then we have a static container-pdu
+                    # we have to loop all sub-pdus and set the offset to the offset of the PDU
+                    # note: order of processing sub-PDUs is not important, even if the sub-PDUs are not ordered
+                    #       by the pdu-offset (we just set the offset correct to the actual processed sub-PDU)
+                    pdu = self.pdus[no_header_next_pdu_idx]
+                    no_header_next_pdu_idx += 1
+                    pdu_dlc = pdu.size
+                    offset = pdu.offset_bytes * 8
+                decode_size_bits = pdu_dlc * 8
                 if pdu is None:
                     return_dict['pdus'].append(None)
                 else:
                     unpacked = self.bitstring_to_signal_list(
                         pdu.signals,
-                        big[offset:offset + pdu_dlc * 8],
-                        little[size - offset - pdu_dlc * 8:size - offset],
-                        pdu_dlc * 8
+                        big[offset:offset + decode_size_bits],
+                        little[size - offset - decode_size_bits:size - offset],
+                        decode_size_bits
                     )
                     pdu_dict = dict()
                     for s, v in zip(pdu.signals, unpacked):
                         pdu_dict[s.name] = DecodedSignal(v, s)
                     return_dict["pdus"].append({pdu.name: pdu_dict})
-                offset += (pdu_dlc * 8)
+                if len(header_signals) > 0:
+                    # if there is a pdu-header, we have to set the offset to the start of the next pdu
+                    offset += decode_size_bits
             return return_dict
         else:
             little, big = self.bytes_to_bitstrings(data)

src/canmatrix/formats/arxml.py

@@ -1110,7 +1110,10 @@ def ar_byteorder_is_little(in_string):
 
 def get_signals(signal_array, frame, ea, multiplex_id, float_factory, bit_offset=0):
     # type: (typing.Sequence[_Element], typing.Union[canmatrix.Frame, canmatrix.Pdu], Earxml, int, typing.Callable, int) -> None
-    """Add signals from xml to the Frame."""
+    """Add signals from xml to the Frame.
+    ATTENTION: be careful if you plan to use bit_offset != 0.
+               This will result in non-valid signal definitions (start-bit) in relation to the PDU.
+               Maybe a possible refactoring in the future to remove this bit_offset if no further need is identified?"""
 
     group_id = 1
     if signal_array is None:  # Empty signalarray - nothing to do
@@ -1493,28 +1496,17 @@ def get_frame_from_container_ipdu(pdu, target_frame, ea, float_factory, headers_
             value = ea.get_child(time_period, "VALUE")
             if value is not None:
                 cycle_time = int(float_factory(value.text) * 1000)
+        # TODO maybe we need to refactor getting the offset for secured-i-pdus to get the offset BEFORE we
+        # copy the authentic/payload-PDU over the ipdu. need to be clarified. For the moment leave it as it was.
+        # until now we have not seen an non-cryptographic secured-i-pdu inside
+        # an static-container-PDU (=without header) (only in that veeeery specific situation it's important)
         try:
-            if header_type == "SHORT-HEADER":
-                header_id = ea.get_child(ipdu, "HEADER-ID-SHORT-HEADER").text
-            elif header_type == "LONG-HEADER":
-                header_id = ea.get_child(ipdu, "HEADER-ID-LONG-HEADER").text
-            else:
-                # none type
-                header_id = None
-        except AttributeError:
-            header_id = None
-        if header_id is not None:
-            header_id = int(header_id, 0)
-
-        if ipdu is not None and 'SECURED-I-PDU' in ipdu.tag:
-            secured_i_pdu_name = ea.get_element_name(ipdu)
-            payload = ea.follow_ref(ipdu, "PAYLOAD-REF")
-            ipdu = ea.follow_ref(payload, "I-PDU-REF")
-            logger.info("found secured pdu '%s', dissolved to '%s'", secured_i_pdu_name, ea.get_element_name(ipdu))
-        try:
-            offset = int(ea.get_child(ipdu, "OFFSET").text, 0) * 8
+            offset_bytes = int(ea.get_child(ipdu, "OFFSET").text, 0)
         except:
-            offset = 0
+            if header_id is None:
+                logger.error(f"PDU {ipdu_name} is a Container-sub-PDU with no header, but NO PDU offset was found! "
+                             f"(will most likely result in wrong encoding/decoding!) - check ARXML file!!")
+            offset_bytes = 0
 
         try:
             pdu_type = ipdu.attrib["DEST"]
@@ -1530,8 +1522,11 @@ def get_frame_from_container_ipdu(pdu, target_frame, ea, float_factory, headers_
         target_pdu = canmatrix.Pdu(name=ipdu_name, size=ipdu_length, id=header_id,
                                    triggering_name=ipdu_triggering_name, pdu_type=pdu_type,
                                    port_type=pdu_port_type, cycle_time=cycle_time)
+                                   port_type=pdu_port_type, cycle_time=cycle_time,
+                                   offset_bytes=offset_bytes)
         pdu_sig_mapping = ea.get_children(ipdu, "I-SIGNAL-TO-I-PDU-MAPPING")
-        get_signals(pdu_sig_mapping, target_pdu, ea, None, float_factory, bit_offset=offset)
+        get_signals(pdu_sig_mapping, target_pdu, ea, None, float_factory, bit_offset=0,
+                    generated_update_bits_init_to_1=generated_update_bits_init_to_1)
         target_frame.add_pdu(target_pdu)