convert: selective rx/tx ecu extraction issue #421 (#422)

Author: ebroecker

docs/cli.rst

@@ -242,6 +242,15 @@ This is some *lite* ECU-Extract.
 
     $ canconvert --ecus=FRONT_ECU,REAR_ECU source.dbc target.dbc
 
+**extract matrix with frames which FRONT_ECU receives and with frames REAR_ECUS transmits:**
+
+::
+
+    $ canconvert --ecus=FRONT_ECU:rx,REAR_ECU:tx source.dbc target.dbc
+
+
+
+
 **extract frame[s] out of matrix:**
 
 ::

src/canmatrix/cli/convert.py

@@ -70,7 +70,7 @@ def get_formats():
 @click.option('--skipLongDlc', 'skipLongDlc', help="skip all Frames with dlc bigger than given threshold")
 @click.option('--cutLongFrames',  'cutLongFrames', help="cut all signals out of Frames with dlc bigger than given threshold")
 @click.option('--deleteFrameAttributes', 'deleteFrameAttributes', help="delete attributes from all frames\nExample --deleteFrameAttributes GenMsgSomeVar,CycleTime")
-@click.option('--ecus', help="Copy only given ECUs (comma separated list) to target matrix")
+@click.option('--ecus', help="Copy only given ECUs (comma separated list) to target matrix; suffix 'rx' or 'tx' for selection: Example: --ecus FirstEcu:rx,SecondEcu:tx,ThirdEcu")
 @click.option('--frames', help="Copy only given Frames (comma separated list) to target matrix")
 @click.option('--signals', help="Copy only given Signals (comma separated list) to target matrix just as 'free' signals without containing frame")
 @click.option('--merge', help="merge additional can databases.\nSyntax: --merge filename[:ecu=SOMEECU][:frame=FRAME1][:frame=FRAME2],filename2")

src/canmatrix/convert.py

@@ -46,20 +46,23 @@ def convert(infile, out_file_name, **options):  # type: (str, str, **str) -> Non
     for name in dbs:
         db = None
 
-        if 'ecus' in options and options['ecus'] is not None:
+        if options.get('ecus', False):
             ecu_list = options['ecus'].split(',')
             db = canmatrix.CanMatrix()
+            direction = None
             for ecu in ecu_list:
-                canmatrix.copy.copy_ecu_with_frames(ecu, dbs[name], db)
-        if 'frames' in options and options['frames'] is not None:
+                if ":" in ecu:
+                    ecu, direction = ecu.split(":")
+                canmatrix.copy.copy_ecu_with_frames(ecu, dbs[name], db, rx=(direction != "tx"), tx=(direction != "rx"))
+        if options.get('frames', False):
             frame_list = options['frames'].split(',')
-            db = canmatrix.CanMatrix()
+            db = canmatrix.CanMatrix() if db is None else db
             for frame_name in frame_list:
                 frame_to_copy = dbs[name].frame_by_name(frame_name)
                 canmatrix.copy.copy_frame(frame_to_copy.arbitration_id, dbs[name], db)
         if options.get('signals', False):
             signal_list = options['signals'].split(',')
-            db = canmatrix.CanMatrix()
+            db = canmatrix.CanMatrix() if db is None else db
             for signal_name in signal_list:
                 canmatrix.copy.copy_signal(signal_name, dbs[name], db)
 

src/canmatrix/copy.py

@@ -65,7 +65,7 @@ def copy_ecu(ecu_or_glob, source_db, target_db):
                     target_db.ecu_defines[attribute].update()
 
 
-def copy_ecu_with_frames(ecu_or_glob, source_db, target_db):
+def copy_ecu_with_frames(ecu_or_glob, source_db, target_db, rx=True, tx=True):
     # type: (typing.Union[canmatrix.Ecu, str], canmatrix.CanMatrix, canmatrix.CanMatrix) -> None
     """
     Copy ECU(s) identified by Name or as Object from source CAN matrix to target CAN matrix.
@@ -87,16 +87,18 @@ def copy_ecu_with_frames(ecu_or_glob, source_db, target_db):
         target_db.add_ecu(copy.deepcopy(ecu))
 
         # copy tx-frames
-        for frame in source_db.frames:
-            if ecu.name in frame.transmitters:
-                copy_frame(frame.arbitration_id, source_db, target_db)
+        if tx is True:
+            for frame in source_db.frames:
+                if ecu.name in frame.transmitters:
+                    copy_frame(frame.arbitration_id, source_db, target_db)
 
         # copy rx-frames
-        for frame in source_db.frames:
-            for signal in frame.signals:
-                if ecu.name in signal.receivers:
-                    copy_frame(frame.arbitration_id, source_db, target_db)
-                    break
+        if rx is True:
+            for frame in source_db.frames:
+                for signal in frame.signals:
+                    if ecu.name in signal.receivers:
+                        copy_frame(frame.arbitration_id, source_db, target_db)
+                        break
 
         # copy all ECU defines
         for attribute in ecu.attributes:

src/canmatrix/tests/test_cli_convert.py

@@ -194,7 +194,7 @@ def test_copy_signals(tmpdir, run):
         assert b"VECTOR__INDEPENDENT_SIG_MSG" in content
 
 
-def create_dbc():
+def create_dbc(additionalReceiver = []):
     outFile = str(here / "tmpb.dbc")
     myFrame = canmatrix.Frame("testFrame3", arbitration_id=canmatrix.arbitration_id_converter(0x124), size=8, transmitters=["testBU"])
     mySignal = canmatrix.Signal("someTestSignal",
@@ -218,7 +218,7 @@ def create_dbc():
                       offset=1.0,
                       min=0,
                       max=500,
-                      receivers=["recBU2"])
+                      receivers=["recBU2"] + additionalReceiver)
     myFrame2.add_signal(mySignal2)
     mySignal3 = canmatrix.Signal("zeroSignal",
                       start_bit=20,
@@ -250,6 +250,22 @@ def test_copy_ecus(tmpdir, run):
         assert "testBU2" not in content
         assert "testBU" in content
 
+def test_copy_ecus_rx(tmpdir, run):
+    inputFile = create_dbc()
+    result = run("--ecus", "recBU:rx", inputFile, "tmp2.dbc")
+    with open("tmp2.dbc","r") as fd:
+        content = fd.read()
+        assert "recBU2" not in content
+        assert "recBU" in content
+
+def test_copy_ecus_tx(tmpdir, run):
+    inputFile = create_dbc(additionalReceiver = ["testBU"])
+    result = run("--ecus", "testBU:tx", inputFile, "tmp2.dbc")
+    with open("tmp2.dbc","r") as fd:
+        content = fd.read()
+        assert "testFrame2" not in content
+        assert "testFrame3" in content
+
 def test_copy_frames(tmpdir, run):
     inputFile = create_dbc()
     result = run("--frames", "testFrame3", inputFile, "tmp2.dbc")