Merge ec68ab2 into 7a83e0a

.pre-commit-config.yaml

@@ -1,10 +1,10 @@
 repos:
   - repo: https://github.com/ambv/black
-    rev: 23.9.1
+    rev: 23.10.0
     hooks:
       - id: black
   - repo: https://github.com/charliermarsh/ruff-pre-commit
     # Ruff version.
-    rev: v0.1.0
+    rev: v0.1.1
     hooks:
       - id: ruff

pyproject.toml

@@ -22,9 +22,15 @@ skip = "pp* *_ppc64le *_s390x *-musllinux* cp312-*"  # skip pypy and irrelevant
 
 [tool.ruff]
 select = [
+    "F",  # pyflakes
     "UP",  # pyupgrade
     "I",  # isort
 ]
+ignore = [
+    "F401",  # unused-import
+    "F821",  # undefined-name
+    "F841",  # unused-variable
+]
 exclude = ["./src/asammdf/gui/ui"]
 target-version = "py38"
 

src/asammdf/blocks/conversion_utils.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 from copy import deepcopy
-from typing import Any, Union
+from typing import Any
 
 from ..types import ChannelConversionType
 from . import v2_v3_blocks as v3b
@@ -147,7 +147,7 @@ def conversion_transfer(
                         conversion.referenced_blocks["default_addr"],
                         v4b.ChannelConversion,
                     ):
-                        default_addr = conversion.referenced_blocks[f"default_addr"].name.encode("latin-1")
+                        default_addr = conversion.referenced_blocks["default_addr"].name.encode("latin-1")
                     else:
                         default_addr = conversion.referenced_blocks["default_addr"]
                     new_conversion.referenced_blocks["default_addr"] = default_addr
@@ -177,7 +177,7 @@ def conversion_transfer(
                         conversion.referenced_blocks["default_addr"],
                         v4b.ChannelConversion,
                     ):
-                        default_addr = conversion.referenced_blocks[f"default_addr"].name.encode("latin-1")
+                        default_addr = conversion.referenced_blocks["default_addr"].name.encode("latin-1")
                     else:
                         default_addr = conversion.referenced_blocks["default_addr"]
                     new_conversion.referenced_blocks["default_addr"] = default_addr
@@ -345,9 +345,9 @@ def from_dict(conversion: dict[str, Any]) -> v4b.ChannelConversion:
 
         val = conversion.get("default_addr", b"")
         if isinstance(val, str):
-            conversion[f"default_addr"] = val.encode("utf-8")
+            conversion["default_addr"] = val.encode("utf-8")
         elif isinstance(val, dict):
-            conversion[f"default_addr"] = from_dict(val)
+            conversion["default_addr"] = from_dict(val)
 
         conversion["ref_param_nr"] = nr + 1
         conversion = v4b.ChannelConversion(**conversion)
@@ -367,9 +367,9 @@ def from_dict(conversion: dict[str, Any]) -> v4b.ChannelConversion:
 
         val = conversion.get("default_addr", b"")
         if isinstance(val, str):
-            conversion[f"default_addr"] = val.encode("utf-8")
+            conversion["default_addr"] = val.encode("utf-8")
         elif isinstance(val, dict):
-            conversion[f"default_addr"] = from_dict(val)
+            conversion["default_addr"] = from_dict(val)
         conversion = v4b.ChannelConversion(**conversion)
 
     else:

src/asammdf/blocks/mdf_v3.py

@@ -46,7 +46,6 @@
 from .. import tool
 from ..signal import Signal
 from ..types import ChannelsType, CompressionType, RasterType, StrPathType
-from ..version import __version__
 from . import v2_v3_constants as v23c
 from .conversion_utils import conversion_transfer
 from .cutils import get_channel_raw_bytes

src/asammdf/blocks/mdf_v4.py

@@ -6,7 +6,7 @@
 
 import bisect
 from collections import defaultdict
-from collections.abc import Callable, Iterable, Iterator, Sequence, Sized
+from collections.abc import Iterator, Sequence
 from copy import deepcopy
 from datetime import datetime
 from functools import lru_cache
@@ -20,11 +20,9 @@
 import shutil
 import sys
 from tempfile import gettempdir, NamedTemporaryFile
-from time import sleep
 from traceback import format_exc
 from typing import Any, overload
 from zipfile import ZIP_DEFLATED, ZipFile
-from zlib import decompress
 
 from typing_extensions import Literal
 
@@ -46,10 +44,8 @@
     array,
     array_equal,
     bool_,
-    bytes_,
     column_stack,
     concatenate,
-    cumsum,
     dtype,
     empty,
     fliplr,
@@ -86,7 +82,6 @@
     StrPathType,
     WritableBufferType,
 )
-from ..version import __version__
 from . import bus_logging_utils
 from . import v4_constants as v4c
 from .conversion_utils import conversion_transfer
@@ -113,7 +108,6 @@
     is_file_like,
     load_can_database,
     MdfException,
-    sanitize_xml,
     SignalDataBlockInfo,
     TERMINATED,
     UINT8_uf,
@@ -148,7 +142,7 @@
 try:
     from isal.isal_zlib import decompress
 except ImportError:
-    pass
+    from zlib import decompress
 
 
 MASTER_CHANNELS = (v4c.CHANNEL_TYPE_MASTER, v4c.CHANNEL_TYPE_VIRTUAL_MASTER)
@@ -177,8 +171,6 @@
     extract,
     get_channel_raw_bytes,
     get_vlsd_max_sample_size,
-    get_vlsd_offsets,
-    lengths,
     sort_data_block,
 )
 
@@ -9496,17 +9488,17 @@ def save(
                                     gp_nr, ch_nr = dep.output_quantity_channel
                                     grp = self.groups[gp_nr]
                                     ch = grp.channels[ch_nr]
-                                    dep[f"output_quantity_dg_addr"] = grp.data_group.address
-                                    dep[f"output_quantity_cg_addr"] = grp.channel_group.address
-                                    dep[f"output_quantity_ch_addr"] = ch.address
+                                    dep["output_quantity_dg_addr"] = grp.data_group.address
+                                    dep["output_quantity_cg_addr"] = grp.channel_group.address
+                                    dep["output_quantity_ch_addr"] = ch.address
 
                                 if dep.comparison_quantity_channel:
                                     gp_nr, ch_nr = dep.comparison_quantity_channel
                                     grp = self.groups[gp_nr]
                                     ch = grp.channels[ch_nr]
-                                    dep[f"comparison_quantity_dg_addr"] = grp.data_group.address
-                                    dep[f"comparison_quantity_cg_addr"] = grp.channel_group.address
-                                    dep[f"comparison_quantity_ch_addr"] = ch.address
+                                    dep["comparison_quantity_dg_addr"] = grp.data_group.address
+                                    dep["comparison_quantity_cg_addr"] = grp.channel_group.address
+                                    dep["comparison_quantity_ch_addr"] = ch.address
 
                                 for i, (gp_nr, ch_nr) in enumerate(dep.axis_channels):
                                     grp = self.groups[gp_nr]
@@ -10314,7 +10306,7 @@ def _process_bus_logging(self) -> None:
                 ):
                     try:
                         self._process_can_logging(index, group)
-                    except Exception as e:
+                    except Exception:
                         message = f"Error during CAN logging processing: {format_exc()}"
                         logger.error(message)
 

src/asammdf/blocks/utils.py

@@ -931,15 +931,15 @@ def count_channel_groups(
                             ch_count += 1
                             ch_addr = UINT64_uf(stream, ch_addr + 24)[0]
                             if ch_addr >= file_limit:
-                                raise MdfException(f"File is a corrupted MDF file - Invalid CH block address")
+                                raise MdfException("File is a corrupted MDF file - Invalid CH block address")
 
                     cg_addr = UINT64_uf(stream, cg_addr + 24)[0]
                     if cg_addr >= file_limit:
-                        raise MdfException(f"File is a corrupted MDF file - Invalid CG block address")
+                        raise MdfException("File is a corrupted MDF file - Invalid CG block address")
 
                 dg_addr = UINT64_uf(stream, dg_addr + 24)[0]
                 if dg_addr >= file_limit:
-                    raise MdfException(f"File is a corrupted MDF file - Invalid DG block address")
+                    raise MdfException("File is a corrupted MDF file - Invalid DG block address")
         else:
             stream.seek(88, 0)
             dg_addr = UINT64_u(stream.read(8))[0]
@@ -956,18 +956,18 @@ def count_channel_groups(
                             stream.seek(ch_addr + 24)
                             ch_addr = UINT64_u(stream.read(8))[0]
                             if ch_addr >= file_limit:
-                                raise MdfException(f"File is a corrupted MDF file - Invalid CH block address")
+                                raise MdfException("File is a corrupted MDF file - Invalid CH block address")
 
                     stream.seek(cg_addr + 24)
                     cg_addr = UINT64_u(stream.read(8))[0]
                     if cg_addr >= file_limit:
-                        raise MdfException(f"File is a corrupted MDF file - Invalid CG block address")
+                        raise MdfException("File is a corrupted MDF file - Invalid CG block address")
 
                 stream.seek(dg_addr + 24)
                 dg_addr = UINT64_u(stream.read(8))[0]
 
                 if dg_addr >= file_limit:
-                    raise MdfException(f"File is a corrupted MDF file - Invalid DG block address")
+                    raise MdfException("File is a corrupted MDF file - Invalid DG block address")
 
     else:
         stream.seek(68, 0)
@@ -985,18 +985,18 @@ def count_channel_groups(
                         stream.seek(ch_addr + 4)
                         ch_addr = UINT32_u(stream.read(4))[0]
                         if ch_addr >= file_limit:
-                            raise MdfException(f"File is a corrupted MDF file - Invalid CH block address")
+                            raise MdfException("File is a corrupted MDF file - Invalid CH block address")
 
                 stream.seek(cg_addr + 4)
                 cg_addr = UINT32_u(stream.read(4))[0]
                 if cg_addr >= file_limit:
-                    raise MdfException(f"File is a corrupted MDF file - Invalid CG block address")
+                    raise MdfException("File is a corrupted MDF file - Invalid CG block address")
 
             stream.seek(dg_addr + 4)
             dg_addr = UINT32_u(stream.read(4))[0]
 
             if dg_addr >= file_limit:
-                raise MdfException(f"File is a corrupted MDF file - Invalid DG block address")
+                raise MdfException("File is a corrupted MDF file - Invalid DG block address")
 
     return count, ch_count
 

src/asammdf/blocks/v2_v3_blocks.py

@@ -27,7 +27,6 @@
 from ..version import __version__
 from . import v2_v3_constants as v23c
 from .utils import (
-    escape_xml_string,
     get_fields,
     get_text_v3,
     MdfException,
@@ -2767,7 +2766,7 @@ def comment(self, string):
             comment = string
             try:
                 comment_xml = ET.fromstring(comment.replace(' xmlns="http://www.asam.net/mdf/v4"', ""))
-            except ET.ParseError as e:
+            except ET.ParseError:
                 self.description = string
             else:
                 description = comment_xml.find(".//TX")

src/asammdf/blocks/v4_blocks.py

@@ -4,9 +4,7 @@
 
 from __future__ import annotations
 
-from bisect import bisect_left, bisect_right
 from datetime import datetime, timedelta, timezone
-from functools import lru_cache
 from hashlib import md5
 import logging
 from pathlib import Path
@@ -52,7 +50,6 @@
     get_text_v4,
     is_file_like,
     MdfException,
-    sanitize_xml,
     UINT8_uf,
     UINT64_u,
     UINT64_uf,
@@ -1483,15 +1480,15 @@ def __init__(self, **kwargs) -> None:
                     links = links[dims_nr * 3 :]
 
                 if self.flags & v4c.FLAG_CA_OUTPUT_QUANTITY:
-                    self[f"output_quantity_dg_addr"] = links[0]
-                    self[f"output_quantity_cg_addr"] = links[1]
-                    self[f"output_quantity_ch_addr"] = links[2]
+                    self["output_quantity_dg_addr"] = links[0]
+                    self["output_quantity_cg_addr"] = links[1]
+                    self["output_quantity_ch_addr"] = links[2]
                     links = links[3:]
 
                 if self.flags & v4c.FLAG_CA_COMPARISON_QUANTITY:
-                    self[f"comparison_quantity_dg_addr"] = links[0]
-                    self[f"comparison_quantity_cg_addr"] = links[1]
-                    self[f"comparison_quantity_ch_addr"] = links[2]
+                    self["comparison_quantity_dg_addr"] = links[0]
+                    self["comparison_quantity_cg_addr"] = links[1]
+                    self["comparison_quantity_ch_addr"] = links[2]
                     links = links[3:]
 
                 if self.flags & v4c.FLAG_CA_AXIS:
@@ -1568,15 +1565,15 @@ def __init__(self, **kwargs) -> None:
                     links = links[dims_nr * 3 :]
 
                 if self.flags & v4c.FLAG_CA_OUTPUT_QUANTITY:
-                    self[f"output_quantity_dg_addr"] = links[0]
-                    self[f"output_quantity_cg_addr"] = links[1]
-                    self[f"output_quantity_ch_addr"] = links[2]
+                    self["output_quantity_dg_addr"] = links[0]
+                    self["output_quantity_cg_addr"] = links[1]
+                    self["output_quantity_ch_addr"] = links[2]
                     links = links[3:]
 
                 if self.flags & v4c.FLAG_CA_COMPARISON_QUANTITY:
-                    self[f"comparison_quantity_dg_addr"] = links[0]
-                    self[f"comparison_quantity_cg_addr"] = links[1]
-                    self[f"comparison_quantity_ch_addr"] = links[2]
+                    self["comparison_quantity_dg_addr"] = links[0]
+                    self["comparison_quantity_cg_addr"] = links[1]
+                    self["comparison_quantity_ch_addr"] = links[2]
                     links = links[3:]
 
                 if self.flags & v4c.FLAG_CA_AXIS:
@@ -1723,16 +1720,16 @@ def __bytes__(self) -> bytes:
 
         if flags & v4c.FLAG_CA_OUTPUT_QUANTITY:
             keys += (
-                f"output_quantity_dg_addr",
-                f"output_quantity_cg_addr",
-                f"output_quantity_ch_addr",
+                "output_quantity_dg_addr",
+                "output_quantity_cg_addr",
+                "output_quantity_ch_addr",
             )
 
         if flags & v4c.FLAG_CA_COMPARISON_QUANTITY:
             keys += (
-                f"comparison_quantity_dg_addr",
-                f"comparison_quantity_cg_addr",
-                f"comparison_quantity_ch_addr",
+                "comparison_quantity_dg_addr",
+                "comparison_quantity_cg_addr",
+                "comparison_quantity_ch_addr",
             )
 
         if flags & v4c.FLAG_CA_AXIS:
@@ -2153,7 +2150,7 @@ def metadata(self) -> str:
 
                     lines[-1] += f" (= '{sep}')"
                 else:
-                    lines[-1] += f" (= <undefined>)"
+                    lines[-1] += " (= <undefined>)"
 
         for line in lines:
             if not line:

src/asammdf/gui/dialogs/advanced_search.py

@@ -5,8 +5,7 @@
 from natsort import natsorted
 from PySide6 import QtCore, QtWidgets
 
-from ...blocks.utils import extract_xml_comment, timeit
-from ..ui import resource_rc
+from ...blocks.utils import extract_xml_comment
 from ..ui.search_dialog import Ui_SearchDialog
 from .messagebox import MessageBox
 from .range_editor import RangeEditor

src/asammdf/gui/dialogs/channel_group_info.py

@@ -1,6 +1,5 @@
 from PySide6 import QtCore, QtGui, QtWidgets
 
-from ..ui import resource_rc
 from ..widgets.channel_group_info import ChannelGroupInfoWidget
 
 

src/asammdf/gui/dialogs/channel_info.py

@@ -1,6 +1,5 @@
 from PySide6 import QtCore, QtGui, QtWidgets
 
-from ..ui import resource_rc
 from ..widgets.channel_info import ChannelInfoWidget