last test for bus tab..

test/asammdf/gui/widgets/batch/test_BatchWidget_Tab_BusLogging.py

@@ -5,10 +5,10 @@
 import urllib.request
 from zipfile import ZipFile
 
+import numpy as np
 import pandas as pd
 from PySide6 import QtCore, QtTest
 
-from asammdf.blocks.utils import load_can_database
 from test.asammdf.gui.test_base import DBC, OpenMDF
 from test.asammdf.gui.widgets.test_BaseBatchWidget import TestBatchWidget
 
@@ -24,45 +24,88 @@ def setUp(self):
         ZipFile(r"test.zip").extractall(self.test_workspace)
         Path("test.zip").unlink()
 
-        # tmp_path = Path("D:\\GHP\\tmp\\asammdf\\BUS")
-        # for file in tmp_path.iterdir():
-        #     shutil.copy(file, self.test_workspace)
-
         temp_dir = Path(self.test_workspace)
 
         # Get test files path
         self.mdf_path = [input_file for input_file in temp_dir.iterdir() if input_file.suffix == ".mf4"][0]
         self.dbc_path = [input_file for input_file in temp_dir.iterdir() if input_file.suffix == ".dbc"][0]
-        npy_path = [input_file for input_file in temp_dir.iterdir() if input_file.suffix == ".npy"]
 
         self.setUpBatchWidget(measurement_files=[str(self.mdf_path)])
         # Go to Tab: "Bus logging": Index 3
         self.widget.aspects.setCurrentIndex(self.bus_aspect)
 
+        # Clear CAN & LIN lists
         self.widget.can_database_list.clear()
         self.widget.lin_database_list.clear()
 
-        # Load DBC
+        # Ensure that CAN & LIN lists are empty
+        self.assertEqual(self.widget.can_database_list.count(), 0)
+        self.assertEqual(self.widget.lin_database_list.count(), 0)
+
+    def load_database(self, path: Path | str | None = None, is_can=True):
+        if not path:
+            path = self.dbc_path
         with mock.patch("asammdf.gui.widgets.batch.QtWidgets.QFileDialog.getOpenFileNames") as mo_getOpenFileNames:
-            mo_getOpenFileNames.return_value = [str(self.dbc_path)], None
-            QtTest.QTest.mouseClick(self.widget.load_can_database_btn, QtCore.Qt.MouseButton.LeftButton)
+            mo_getOpenFileNames.return_value = [str(path)], None
+            if is_can:
+                QtTest.QTest.mouseClick(self.widget.load_can_database_btn, QtCore.Qt.MouseButton.LeftButton)
+            else:
+                QtTest.QTest.mouseClick(self.widget.load_lin_database_btn, QtCore.Qt.MouseButton.LeftButton)
 
+    def test_load_can_database_btn(self):
+        """
+        Events:
+            - Press Load CAN database button.
+
+        Evaluate:
+            - There is one item in can database list
+            - The item's text is .dbc path
+        """
+        # Event
+        self.load_database()
+        # Evaluate
         self.assertEqual(self.widget.can_database_list.count(), 1)
-        self.can_database_matrix = load_can_database(self.dbc_path)
+        self.assertEqual(
+            self.widget.can_database_list.itemWidget(self.widget.can_database_list.item(0)).database.text(),
+            str(self.dbc_path),
+        )
 
-    def test_extract_bus_btn(self):
+    def test_load_lin_database_btn(self):
         """
+        Events:
+            - Press Load LIN database button.
 
-        Returns
-        -------
+        Evaluate:
+            - There is one item in lin database list
+            - The item's text is .dbc path
+        """
+        # Event
+        self.load_database(is_can=False)
+        # Evaluate
+        self.assertEqual(self.widget.lin_database_list.count(), 1)
+        self.assertEqual(
+            self.widget.lin_database_list.itemWidget(self.widget.lin_database_list.item(0)).database.text(),
+            str(self.dbc_path),
+        )
 
+    def test_extract_bus_btn(self):
+        """
+        When QThreads are running, event-loops needs to be processed.
+        Events:
+            - Set prefix text
+            - Press PushButton Extract Bus signals.
+
+        Evaluate:
+            - File was created.
+            - New channels was created from .dbc and .mf4 files
         """
         # Expected result
         output_file = Path.with_suffix(self.mdf_path, f".bus_logging{self.mdf_path.suffix}")
 
         # Precondition
+        self.load_database()
         self.assertFalse(output_file.exists())
-        self.assertEqual(self.widget.output_info_bus.toPlainText(), "")
+        self.assertEqual(self.widget.output_info_bus.toPlainText(), "")  # bus output info tab is clean
 
         # Set Prefix
         self.widget.prefix.setText(self.id().split(".")[-1])
@@ -130,16 +173,25 @@ def test_extract_bus_btn(self):
             self.assertEqual(max(timestamps_max), max_from_mdf)
 
     def test_extract_bus_csv_btn_0(self):
-        """
+        """ """
+        # Prepare expected results
+        with OpenMDF(self.mdf_path) as mdf_file, open(self.dbc_path) as dbc_file:
+            for key, value in mdf_file.bus_logging_map.items():
+                if value:
+                    prefix = key
+                    for new_key, new_value in value.items():
+                        if value:
+                            prefix += str(new_key)
+            self.assertTrue(prefix)  # there is a suffix for feature channels
+            prefix += "." + DBC.BO(dbc_file.readlines()).name + "."
+        to_replace = [" ", '"', ":"]
 
-        Returns
-        -------
-
-        """
         # Precondition
+        self.load_database()
         for file in Path(self.test_workspace).iterdir():
             self.assertNotEqual(file.suffix, ".csv")
         self.assertEqual(self.widget.output_info_bus.toPlainText(), "")
+        self.toggle_checkboxes(widget=self.widget.extract_bus_tab, check=False)  # uncheck all checkboxes
 
         # Expected results
         output_file = Path.with_suffix(self.mdf_path, f".bus_logging.mf4")
@@ -148,13 +200,123 @@ def test_extract_bus_csv_btn_0(self):
         self.widget.prefix.setText(self.id().split(".")[-1])
 
         # Get new mdf file
-        self.mouse_click_on_btn_with_progress(self.widget.extract_csv_btn)
+        self.mouse_click_on_btn_with_progress(self.widget.extract_bus_btn)
 
         # Event
         self.mouse_click_on_btn_with_progress(self.widget.extract_bus_csv_btn)
 
-        csv_tables = {file: pd.read_csv(file) for file in Path(self.test_workspace).iterdir() if file.suffix == ".csv"}
+        csv_tables = [
+            (file, pd.read_csv(file)) for file in Path(self.test_workspace).iterdir() if file.suffix == ".csv"
+        ]
         with OpenMDF(output_file) as mdf_file:
+            for group, (path, table) in zip(mdf_file.groups, csv_tables):
+                comment = group.channel_group.comment
+                for char in to_replace:
+                    comment = comment.replace(char, "_")
+
+                # Evaluate CSV name
+                self.assertIn(self.id().split(".")[-1], str(path))
+                self.assertTrue(str(path).endswith(comment + ".csv"))
+                self.assertTrue(str(path.stem).startswith(output_file.stem))
+
+                # Evaluate channels
+                for channel in group.channels:
+                    if channel.name != "time":
+                        name = prefix + channel.name
+                        self.assertIn(name, table.columns)
+                        ch = mdf_file.get(channel.name)
+
+                        # Evaluate samples
+                        self.assertEqual(ch.samples.size, table[name].size)
+                        if np.issubdtype(ch.samples.dtype, np.number):
+                            self.assertAlmostEqual(ch.samples.min(), table[name].min(), places=10)
+                            self.assertAlmostEqual(ch.samples.max(), table[name].max(), places=10)
+                        else:
+                            self.assertEqual(
+                                str(min(ch.samples)).replace(" ", ""),
+                                str(table[name].min()).replace(" ", ""),
+                            )
+                            self.assertEqual(
+                                str(max(ch.samples)).replace(" ", ""),
+                                str(table[name].max()).replace(" ", ""),
+                            )
+
+                        # Evaluate timestamps
+                        self.assertEqual(ch.timestamps.size, table.timestamps.size)
+                        self.assertAlmostEqual(ch.timestamps.min(), table.timestamps.min(), places=10)
+                        self.assertAlmostEqual(ch.timestamps.max(), table.timestamps.max(), places=10)
+
+    def test_extract_bus_csv_btn_1(self):
+        """ """
+        # Prepare expected results
+        with OpenMDF(self.mdf_path) as mdf_file, open(self.dbc_path) as dbc_file:
+            for key, value in mdf_file.bus_logging_map.items():
+                if value:
+                    prefix = key
+                    for new_key, new_value in value.items():
+                        if value:
+                            prefix += str(new_key)
+            self.assertTrue(prefix)  # there is a suffix for feature channels
+            prefix += "." + DBC.BO(dbc_file.readlines()).name + "."
+        to_replace = [" ", '"', ":"]
+
+        # Precondition
+        self.load_database()
+        for file in Path(self.test_workspace).iterdir():
+            self.assertNotEqual(file.suffix, ".csv")
+        self.assertEqual(self.widget.output_info_bus.toPlainText(), "")
+        self.toggle_checkboxes(widget=self.widget.extract_bus_tab, check=True)  # uncheck all checkboxes
 
-            print("ok")
-            x = 5
+        # Expected results
+        csv_path = Path.with_suffix(self.mdf_path, f".bus_logging.csv")
+        output_file = Path.with_suffix(self.mdf_path, f".bus_logging.mf4")
+
+        # Set Prefix
+        self.widget.prefix.setText(self.id().split(".")[-1])
+
+        # Get new mdf file
+        self.mouse_click_on_btn_with_progress(self.widget.extract_bus_btn)
+
+        # Event
+        self.mouse_click_on_btn_with_progress(self.widget.extract_bus_csv_btn)
+
+        csv_table = pd.read_csv(csv_path, header=[0, 1], engine="python", encoding="ISO-8859-1")
+        with OpenMDF(output_file) as mdf_file:
+            # Get channels timestamps max, min, difference between extremes
+            min_ts = min(channel.timestamps.min() for channel in mdf_file.iter_channels())
+            max_ts = max(channel.timestamps.max() for channel in mdf_file.iter_channels())
+            seconds = int(max_ts - min_ts)
+            microseconds = np.floor((max_ts - min_ts - seconds) * pow(10, 6))
+
+            delay = np.datetime64(mdf_file.start_time) - np.timedelta64(3, "h")
+            csv_table.timestamps = pd.DatetimeIndex(csv_table.timestamps.values, yearfirst=True).values - delay
+
+            # Evaluate timestamps min
+            self.assertEqual(np.timedelta64(csv_table.timestamps.values.min(), "us").item().microseconds, 0)
+            self.assertEqual(np.timedelta64(csv_table.timestamps.values.min(), "us").item().seconds, 0)
+            # Evaluate timestamps max
+            self.assertEqual(
+                np.timedelta64(csv_table.timestamps.values.max(), "us").item().microseconds, int(microseconds)
+            )
+            self.assertEqual(np.timedelta64(csv_table.timestamps.values.max(), "us").item().seconds, seconds)
+
+            for ch in mdf_file.iter_channels():
+                name = prefix + ch.name
+                self.assertIn(name, csv_table.columns)
+                column = csv_table[name]
+                if ch.unit:
+                    self.assertEqual(ch.unit, column.columns[0])
+
+                # Evaluate channel samples
+                if np.issubdtype(ch.samples.dtype, np.number):
+                    self.assertAlmostEqual(ch.samples.min(), column.values.min(), places=10)
+                    self.assertAlmostEqual(ch.samples.max(), column.values.max(), places=10)
+                else:
+                    self.assertEqual(
+                        str(min(ch.samples)).replace(" ", ""),
+                        str(column.values.min()).replace(" ", ""),
+                    )
+                    self.assertEqual(
+                        str(max(ch.samples)).replace(" ", ""),
+                        str(column.values.max()).replace(" ", ""),
+                    )

test/asammdf/gui/widgets/batch/test_BatchWidget_Tab_ModifyAndExport.py

@@ -1,9 +1,9 @@
 #!/usr/bin/env python
 import datetime
+import unittest
 from math import ceil
 from pathlib import Path
 from random import randint
-import unittest
 from unittest import mock
 import urllib
 import urllib.request
@@ -115,19 +115,14 @@ def generic_setup(self, name: str = "MDF", check: bool = True):  # Parquet is re
         """
         self.widget.files_list.addItems([str(Path(self.resource, self.default_test_file))])
         self.select_random_channels()
-        self.processEvents()
 
-        for q_widget in self.widget.output_options.children():
+        for q_widget in self.widget.output_format.children():
             if name in q_widget.objectName():
                 break
         else:
             unittest.skip(f"{name} output option widget is not defined")
-        # set checkboxes check state
-        for checkbox in q_widget.findChildren(QtWidgets.QCheckBox):
-            if checkbox.isChecked() != check:
-                self.mouseClick_CheckboxButton(checkbox)
-        if not checkbox:
-            unittest.skip(f"{name} output option widget does not have checkboxes")
+
+        self.toggle_checkboxes(widget=self.widget.output_options, check=check)
 
         self.processEvents(0.01)
 

test/asammdf/gui/widgets/test_BaseBatchWidget.py

@@ -1,12 +1,13 @@
+import unittest
 from collections.abc import Iterable, Sequence
 import os
 from pathlib import Path
 import shutil
 from unittest import mock
 
-from PySide6 import QtCore
+from PySide6 import QtCore, QtWidgets
 from PySide6.QtTest import QTest
-from PySide6.QtWidgets import QPushButton, QTreeWidgetItemIterator
+from PySide6.QtWidgets import QPushButton, QTreeWidgetItemIterator, QWidget
 
 from asammdf.gui.widgets.batch import BatchWidget
 from test.asammdf.gui.test_base import TestBase
@@ -123,3 +124,9 @@ def mouse_click_on_btn_with_progress(self, btn: QPushButton):
         while self.widget._progress:
             self.processEvents(0.01)
         self.processEvents()
+
+    def toggle_checkboxes(self, widget: QWidget, check=True):
+        # set checkboxes check state
+        for checkbox in widget.findChildren(QtWidgets.QCheckBox):
+            if checkbox.isChecked() != check:
+                self.mouseClick_CheckboxButton(checkbox)

test/requirements.txt

@@ -2,21 +2,4 @@ PyAutoGUI
 pytest
 pytest-cov
 pytest-timeout
-xmlrunner; python_version < "3.12"
-
-PySide6==6.7.3
-canmatrix[arxml,dbc]>=1.0
-h5py
-isal; platform_machine == "x86_64" or platform_machine == "AMD64"
-lz4
-natsort
-numexpr
-numpy<2.0.0
-pandas
-psutil
-pyautogui
-pyqtgraph
-python-dateutil
-python-xlib; sys_platform=="linux"
-scipy
-typing_extensions
+xmlrunner; python_version < "3.12"