Merge pull request #19 from Rohde-Schwarz/18-npz-loader-and-saver

Adding npz file format loader and saver

src/RsWaveform/__init__.py

@@ -9,10 +9,10 @@
 import numpy as np
 
 from . import iqtar, iqw, wv
-from .LoadInterface import LoadInterface
+from .load_interface import LoadInterface
 from .meta import Meta
-from .ParentStorage import ParentStorage
-from .SaveInterface import SaveInterface
+from .parent_storage import ParentStorage
+from .save_interface import SaveInterface
 from .utility.dsp import (
     calculate_par,
     calculate_peak,
@@ -192,8 +192,8 @@ def save(
     "RsWaveform",
     "Iqw",
     "IqTar",
-    "LoadInterface",
-    "SaveInterface",
+    "load_interface",
+    "save_interface",
     "calculate_par",
     "calculate_peak",
     "calculate_rms",

src/RsWaveform/iqtar/__init__.py

@@ -1,6 +1,6 @@
 """Iq tar loader and saver implementation."""
 
-from .Load import Load
-from .Save import Save
+from .load import Load
+from .save import Save
 
 __all__ = ["Load", "Save"]

src/RsWaveform/iqtar/Load.py → src/RsWaveform/iqtar/load.py

@@ -13,9 +13,9 @@
 from RsWaveform.meta import Meta
 
 from ..iqw import Load as LoadIqw
-from ..LoadInterface import LoadInterface
-from ..ParentStorage import ParentStorage
-from ..Storage import Storage
+from ..load_interface import LoadInterface
+from ..parent_storage import ParentStorage
+from ..storage import Storage
 from ..utility.file_handling import read_file_handle, read_file_handle_tar
 from ..utility.meta_utils import map_meta_information_name
 

src/RsWaveform/iqtar/Save.py → src/RsWaveform/iqtar/save.py

@@ -5,11 +5,11 @@
 import typing
 from pathlib import Path
 
-from ..iqw.Save import Save as SaveIqw
+from ..iqw.save import Save as SaveIqw
 from ..meta.defaults import META_IQTAR_DEFAULTS as META_DEFAULTS
-from ..ParentStorage import ParentStorage
-from ..SaveInterface import SaveInterface
-from ..Storage import Storage
+from ..parent_storage import ParentStorage
+from ..save_interface import SaveInterface
+from ..storage import Storage
 from ..utility.file_handling import write_file_handle, write_file_handle_tar
 
 

src/RsWaveform/iqw/__init__.py

@@ -1,6 +1,6 @@
 """Iqw loader and saver implementation."""
 
-from .Load import Load
-from .Save import Save
+from .load import Load
+from .save import Save
 
 __all__ = ["Load", "Save"]

src/RsWaveform/iqw/Load.py → src/RsWaveform/iqw/load.py

@@ -6,8 +6,8 @@
 
 import numpy as np
 
-from ..LoadInterface import LoadInterface
-from ..ParentStorage import ParentStorage
+from ..load_interface import LoadInterface
+from ..parent_storage import ParentStorage
 from ..utility.fake_jit import jit
 from ..utility.file_handling import read_file_handle
 

src/RsWaveform/iqw/Save.py → src/RsWaveform/iqw/save.py

@@ -7,9 +7,9 @@
 
 import numpy as np
 
-from ..ParentStorage import ParentStorage
-from ..SaveInterface import SaveInterface
-from ..Storage import Storage
+from ..parent_storage import ParentStorage
+from ..save_interface import SaveInterface
+from ..storage import Storage
 from ..utility.fake_jit import jit
 from ..utility.file_handling import write_file_handle
 

src/RsWaveform/LoadInterface.py → src/RsWaveform/load_interface.py

@@ -5,7 +5,7 @@
 import abc
 import typing
 
-from .ParentStorage import ParentStorage
+from .parent_storage import ParentStorage
 
 if typing.TYPE_CHECKING:
     from pathlib import Path

src/RsWaveform/npz/README.md

@@ -0,0 +1,4 @@
+# The Npz loader and saver
+
+The Npz loader and saver comes also with an extended interface providing a `dtype` parameter
+for setting the format of the output `i` and `q` values accordingly for loading and saving.

src/RsWaveform/npz/__init__.py

@@ -0,0 +1,6 @@
+"""Npz waveform loader and saver implementation."""
+
+from .load import Load
+from .save import Save
+
+__all__ = ["Load", "Save"]

src/RsWaveform/npz/load.py

@@ -0,0 +1,90 @@
+"""Load implementation for generic waveform."""
+
+from __future__ import annotations
+
+import logging
+import typing
+
+import numpy as np
+
+from ..load_interface import LoadInterface
+from ..meta import Meta
+from ..parent_storage import ParentStorage
+from ..storage import Storage
+from .npz_inteface import NpzInterface
+
+if typing.TYPE_CHECKING:
+    from pathlib import Path
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class Load(LoadInterface, NpzInterface):
+    """Load class for generic waveform."""
+
+    def load(self, file: typing.Union[str, typing.IO, Path]) -> ParentStorage:
+        """Load waveform data from npz file."""
+        content = np.load(file=file, allow_pickle=True)
+        data_objs = content.get("storages")
+        if not data_objs:
+            raise ValueError(f"File {file} is not compliant to data schema.")
+        parent = ParentStorage(number_of_storages=len(data_objs))
+        for idx, obj in enumerate(data_objs):
+            i_values = obj.get("i")
+            q_values = obj.get("q")
+            meta = obj.get("meta", {})
+            if i_values is None:
+                LOGGER.warning(
+                    (
+                        "Skipping index %s of file %s content because no i values are"
+                        " in it."
+                    ),
+                    idx,
+                    str(file),
+                )
+                continue
+            if q_values is None:
+                q_values = np.zeros(i_values.size)
+            if q_values is not None and len(i_values) != len(q_values):
+                LOGGER.warning(
+                    (
+                        "Skipping index %s of file %s content because i value and"
+                        " q value lengths are not identical."
+                    ),
+                    idx,
+                    str(file),
+                )
+                continue
+            storage = Storage()
+            storage.data = i_values.astype(self.dtype) + 1j * q_values.astype(
+                self.dtype
+            )
+            storage.meta = Meta(no_defaults=True, **meta)
+            parent.storages[idx] = storage
+        return parent
+
+    def load_in_chunks(
+        self,
+        file: typing.Union[str, typing.IO, Path],
+        samples: int,
+        offset: int,
+    ) -> ParentStorage:
+        """Load chunk waveform data from file."""
+        parent = self.load(file)
+        for idx, storage in enumerate(parent.storages):
+            if offset + samples > len(storage.data):
+                raise ValueError(
+                    f"Could not load chunks for index {idx} of file {file} because "
+                    f"offset {offset} and sample count {samples} is greater than length"
+                    f" of available data {len(storage.data)}. "
+                )
+            storage.data = storage.data[offset : offset + samples]
+        return parent
+
+    def load_meta(self, file: typing.Union[str, typing.IO, Path]) -> ParentStorage:
+        """Load meta information only from wv file."""
+        parent = self.load(file)
+        for storage in parent.storages:
+            storage.data = np.empty((0, 0))
+        return parent

src/RsWaveform/npz/npz_inteface.py

@@ -0,0 +1,30 @@
+"""Npz interface."""
+
+import numpy as np
+import numpy.typing as npt
+
+
+class NpzInterface:
+    """Interface class for Npz."""
+
+    def __init__(self) -> None:
+        """Instantiate interface class."""
+        self._dtype: npt.DTypeLike = np.dtype(np.float16)
+
+    @property
+    def dtype(self) -> npt.DTypeLike:
+        """Dtype for loading and saving.
+
+        Returns:
+            DTypeLike: Numpy dtype
+        """
+        return self._dtype
+
+    @dtype.setter
+    def dtype(self, value: npt.DTypeLike) -> None:
+        """Dtype for loading and saving.
+
+        Args:
+            value (DTypeLike): Numpy dtype
+        """
+        self._dtype = value

src/RsWaveform/npz/save.py

@@ -0,0 +1,33 @@
+"""Save implementation for generic waveform."""
+
+from __future__ import annotations
+
+import typing
+
+import numpy as np
+
+from ..parent_storage import ParentStorage
+from ..save_interface import SaveInterface
+from .npz_inteface import NpzInterface
+
+if typing.TYPE_CHECKING:
+    from pathlib import Path
+
+
+class Save(SaveInterface, NpzInterface):
+    """Save class for generic waveform."""
+
+    def save(
+        self,
+        file: typing.Union[str, typing.IO, Path],
+        datas: ParentStorage,
+        scale: float = np.power(2, np.iinfo(np.int16).bits - 1),
+    ) -> None:
+        """Save waveform data to file."""
+        data_objs = []
+        for storage in datas.storages:
+            i_values = np.real(storage.data).astype(self.dtype)
+            q_values = np.imag(storage.data).astype(self.dtype)
+            meta = storage.meta._items  # pylint: disable=W0212
+            data_objs.append({"i": i_values, "q": q_values, "meta": meta})
+        np.savez_compressed(file=file, storages=data_objs)  # type: ignore

src/RsWaveform/ParentStorage.py → src/RsWaveform/parent_storage.py

@@ -4,7 +4,7 @@
 import typing
 from datetime import datetime
 
-from .Storage import Storage
+from .storage import Storage
 
 
 class ParentStorage(abc.ABC):

src/RsWaveform/SaveInterface.py → src/RsWaveform/save_interface.py

@@ -5,7 +5,7 @@
 import abc
 import typing
 
-from .ParentStorage import ParentStorage
+from .parent_storage import ParentStorage
 
 if typing.TYPE_CHECKING:
     from pathlib import Path

src/RsWaveform/wv/__init__.py

@@ -1,6 +1,6 @@
 """Signal Generator Waveform loader and saver implementation."""
 
-from .Load import Load
-from .Save import Save
+from .load import Load
+from .save import Save
 
 __all__ = ["Load", "Save"]

src/RsWaveform/wv/Load.py → src/RsWaveform/wv/load.py

@@ -9,10 +9,10 @@
 
 import numpy as np
 
-from ..LoadInterface import LoadInterface
+from ..load_interface import LoadInterface
 from ..meta import Meta
-from ..ParentStorage import ParentStorage
-from ..Storage import Storage
+from ..parent_storage import ParentStorage
+from ..storage import Storage
 from ..utility.fake_jit import jit
 from ..utility.file_handling import read_file_handle
 from .utility.array_to_bytes import unpack_bytes_to_bool_array
@@ -92,20 +92,20 @@ def _read_chunks(stream: typing.IO, separators: list) -> typing.Tuple[bytes, byt
     def load_in_chunks(
         self,
         file: typing.Union[str, typing.IO, Path],
-        nr_samples: int,
-        samples_offset: int,
+        samples: int,
+        offset: int,
     ) -> ParentStorage:
         """Load chunk waveform data from file."""
         separators = ["{WWAVEFORM", "{WAVEFORM"]
         with read_file_handle(file) as fp:
             header_content, content = self._read_chunks(fp, separators)
-            content += fp.read((nr_samples + 4) * 4 + 100)
+            content += fp.read((samples + 4) * 4 + 100)
             # +4 due to opt words and 100 just an offset to make sure we get everything
 
         tags = self._split_data_via_tags_meta(header_content)
         tags.update(self._split_data_via_tags_waveform(content, True))
         tags.update(
-            self._split_data_via_tags_control_list_width4(content, nr_samples, True)
+            self._split_data_via_tags_control_list_width4(content, samples, True)
         )
 
         mwv_segment_count = int(tags.pop("mwv segment count", 1))
@@ -115,13 +115,13 @@ def load_in_chunks(
         if isinstance(file, str):
             parent_storage.filename = file
         storages = [Storage() for _ in range(mwv_segment_count)]
-        iq_data = self._extract_iq_chunks(tags, nr_samples, samples_offset)
+        iq_data = self._extract_iq_chunks(tags, samples, offset)
         index = 0
         storages[index].data = iq_data
         meta = self._extract_meta(tags, index)
         meta = self._handle_mwv_meta_data(meta, index)
         storages[index].meta = Meta(**meta)
-        if nr_samples != len(iq_data):
+        if samples != len(iq_data):
             raise ValueError(
                 "Sanity problem. Sample setting and actual sample count does not match."
             )

src/RsWaveform/wv/Save.py → src/RsWaveform/wv/save.py

@@ -8,9 +8,9 @@
 import numpy as np
 
 from ..meta.defaults import META_WV_DEFAULTS as META_DEFAULTS
-from ..ParentStorage import ParentStorage
-from ..SaveInterface import SaveInterface
-from ..Storage import Storage
+from ..parent_storage import ParentStorage
+from ..save_interface import SaveInterface
+from ..storage import Storage
 from ..utility.dsp import calculate_peak, calculate_rms
 from ..utility.fake_jit import jit
 from ..utility.file_handling import write_file_handle

tests/conftest.py

@@ -49,3 +49,9 @@ def reference_iqtar_file_name_huge() -> str:
 def reference_iqtar_file_name_multi_channel() -> str:
     dirname = os.path.dirname(__file__)
     return os.path.join(os.path.realpath(dirname), "data", "dummy_multi_channel.iq.tar")
+
+
+@pytest.fixture
+def reference_npz_file_name() -> str:
+    dirname = os.path.dirname(__file__)
+    return os.path.join(os.path.realpath(dirname), "data", "dummy.npz")

tests/test_Storage.py

@@ -12,8 +12,8 @@
 import pytest
 
 from RsWaveform.meta import Meta
-from RsWaveform.ParentStorage import ParentStorage
-from RsWaveform.Storage import Storage, encode_datetime, msgpack_import
+from RsWaveform.parent_storage import ParentStorage
+from RsWaveform.storage import Storage, encode_datetime, msgpack_import
 
 
 @pytest.fixture