prove of concept for flexray dump (#432)

Author: ebroecker

examples/fr_dump.py

@@ -0,0 +1,14 @@
+#!/usr/bin/env python3
+import canmatrix.formats
+import sys
+cluster = canmatrix.formats.loadp(sys.argv[1], decode_flexray = True)
+
+for cm in cluster:
+    for frame in cluster[cm]:
+        frame_info = "{}-{}-{}".format(frame.slot_id, frame.base_cycle, frame.repitition_cycle)
+        for pdu in frame.pdus:
+            for signal in pdu.signals:
+                sig_group = pdu.get_signal_group_for_signal(signal)
+                sig_group = "None" if sig_group is None else sig_group.name
+                print("{}: {}, {}, {}, {}, {}".format(frame_info, frame.size, pdu.pdu_type, pdu.name, sig_group, signal.name))
+

src/canmatrix/canmatrix.py

@@ -721,6 +721,73 @@ def __eq__(self, other):
             )
         )
 
+@attr.s(cmp=False)
+class Pdu(object):
+    """
+    Represents a PDU.
+
+    PDUs are hierarchical groups of signals which are needed to represent Flexray busses
+    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.
+    """
+
+    name = attr.ib(default="")  # type: str
+    size = attr.ib(default=0)  # type: int
+    triggering_name = attr.ib(default="")  # type: str
+    pdu_type = attr.ib(default="")  # type: str
+    port_type = attr.ib(default="")  # type: str
+    signals = attr.ib(factory=list)  # type: typing.MutableSequence[Signal]
+    signalGroups = attr.ib(factory=list)  # type: typing.MutableSequence[SignalGroup]
+
+    def add_signal(self, signal):
+        # type: (Signal) -> Signal
+        """
+        Add Signal to Pdu.
+
+        :param Signal signal: Signal to be added.
+        :return: the signal added.
+        """
+        self.signals.append(signal)
+        return self.signals[len(self.signals) - 1]
+    def add_signal_group(self, Name, Id, signalNames):
+        # type: (str, int, typing.Sequence[str]) -> None
+        """Add new SignalGroup to the Frame. Add given signals to the group.
+
+        :param str Name: Group name
+        :param int Id: Group id
+        :param list of str signalNames: list of Signal names to add. Non existing names are ignored.
+        """
+        newGroup = SignalGroup(Name, Id)
+        self.signalGroups.append(newGroup)
+        for signal in signalNames:
+            signal = signal.strip()
+            if signal.__len__() == 0:
+                continue
+            signalId = self.signal_by_name(signal)
+            if signalId is not None:
+                newGroup.add_signal(signalId)
+
+    def get_signal_group_for_signal(self, signal_to_find):
+        for signal_group in self.signalGroups:
+            for signal in signal_group:
+                if signal == signal_to_find:
+                    return signal_group
+        return None
+
+    def signal_by_name(self, name):
+        # type: (str) -> typing.Union[Signal, None]
+        """
+        Get signal by name.
+
+        :param str name: signal name to be found.
+        :return: signal with given name or None if not found
+        """
+        for signal in self.signals:
+            if signal.name == name:
+                return signal
+        return None
+
 
 @attr.s(cmp=False)
 class Frame(object):
@@ -764,6 +831,9 @@ class Frame(object):
     # ('cycleTime', '_cycleTime', int, None),
     # ('sendType', '_sendType', str, None),
 
+    pdus = attr.ib(factory=list)  # type: typing.MutableSequence[Pdu]
+
+
     @property
     def is_multiplexed(self):  # type: () -> bool
         """Frame is multiplexed if at least one of its signals is a multiplexer."""
@@ -879,6 +949,7 @@ def attribute(self, attribute_name, db=None, default=None):
 
     def __iter__(self):  # type: () -> typing.Iterator[Signal]
         """Iterator over all signals."""
+
         return iter(self.signals)
 
     def add_signal_group(self, Name, Id, signalNames):
@@ -912,6 +983,18 @@ def signal_group_by_name(self, name):
                 return signalGroup
         return None
 
+    def add_pdu(self, pdu):
+        # type: (Pdu) -> Pdu
+        """
+        Add Pdu to Frame.
+
+        :param Pdu pdu: Pdu to be added.
+        :return: the pdu added.
+        """
+        self.pdus.append(pdu)
+        return self.pdus[len(self.pdus) - 1]
+
+
     def add_signal(self, signal):
         # type: (Signal) -> Signal
         """

src/canmatrix/formats/arxml.py

@@ -1631,30 +1631,37 @@ def decode_flexray_helper(root, root_or_cache, ns, float_factory):
         physical_channels = fc.findall('.//' + ns + "FLEXRAY-PHYSICAL-CHANNEL")
         for pc in physical_channels:
             db = canmatrix.CanMatrix()
+            db.is_flexray = True
             db.add_signal_defines("LongName", 'STRING')
             channel_name = get_element_name(pc, ns)
             found_matrixes[channel_name] = db
 
             frames = pc.findall('.//' + ns + "FLEXRAY-FRAME-TRIGGERING")
             for frame in frames:
+                frame_counter += 1
                 slot_id = int(get_child(frame, "SLOT-ID", root_or_cache, ns).text)
-                base_cycle = get_child(frame, "BASE-CYCLE", root_or_cache, ns)
+                base_cycle = get_child(frame, "BASE-CYCLE", root_or_cache, ns).text
                 ipdu_triggerings = get_children(frame, "I-PDU-TRIGGERING", root_or_cache, ns)
-                frame_repetition_cycle = find_children_by_path(frame, "CYCLE-REPETITION/CYCLE-REPETITION", root_or_cache, ns)[0]
+                frame_repetition_cycle = find_children_by_path(frame, "CYCLE-REPETITION/CYCLE-REPETITION", root_or_cache, ns)[0].text
                 network_endpoints = pc.findall('.//' + ns + "NETWORK-ENDPOINT")
                 frame_size = int(find_children_by_path(frame, "FRAME/FRAME-LENGTH", root_or_cache, ns)[0].text)
+                frame = canmatrix.Frame(size = frame_size, arbitration_id = frame_counter)
+                frame.slot_id = slot_id
+                frame.base_cycle = base_cycle
+                frame.repitition_cycle = frame_repetition_cycle.replace("CYCLE-REPETITION-","")
+                db.add_frame(frame)
                 for ipdu_triggering in ipdu_triggerings:
                     ipdu_triggering_name = get_element_name(ipdu_triggering, ns)
                     ipdu = get_child(ipdu_triggering, "I-PDU", root_or_cache, ns)
                     pdu_type = get_child(ipdu_triggering, "I-PDU-REF", root_or_cache, ns).attrib["DEST"]
                     ipdu_length = int(get_child(ipdu, "LENGTH", root_or_cache, ns).text)
                     pdu_port_type = get_child(ipdu_triggering, "I-PDU-PORT-REF", root_or_cache, ns).attrib["DEST"]
                     ipdu_name = get_element_name(ipdu, ns)
-                    frame_counter += 1;
-                    target_frame = canmatrix.Frame(name = ipdu_name, arbitration_id=frame_counter, size=ipdu_length)
+                    target_pdu = canmatrix.Pdu(name = ipdu_name, size=ipdu_length,
+                                               triggering_name = ipdu_triggering_name, pdu_type=pdu_type, port_type=pdu_port_type)
                     pdu_sig_mapping = get_children(ipdu, "I-SIGNAL-TO-I-PDU-MAPPING", root_or_cache, ns)
-                    get_signals(pdu_sig_mapping, target_frame, root_or_cache, ns, None, float_factory)
-                    db.add_frame(target_frame)
+                    get_signals(pdu_sig_mapping, target_pdu, root_or_cache, ns, None, float_factory)
+                    frame.add_pdu(target_pdu)
     return found_matrixes
 
 def decode_can_helper(root, root_or_cache, ns, float_factory, ignore_cluster_info):