Zyq/writer struct

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pystdf4"
-version = "0.1.0"
+version = "0.1.1"
 description = "A Python library for reading and writing STDF4 files"
 readme = "README.md"
 requires-python = ">=3.8"
@@ -23,7 +23,5 @@ docstring-code-format = true
 [dependency-groups]
 dev = [
  "coveralls>=4.0.1",
- "cython>=3.2.1",
  "pytest>=8.3.5",
- "setuptools>=75.3.2",
 ]

pystdf4/Core/__init__.py

@@ -1,31 +1,22 @@
-from .data_type import (
-    B_1,
-    C_1,
-    I_1,
-    I_2,
-    I_4,
-    R_4,
-    R_8,
-    U_1,
-    U_2,
-    U_4,
-    B_n,
-    C_n,
-)
+from .data_type import B_1, C_1, I_1, I_2, I_4, R_4, R_8, U_1, U_2, U_4, B_n, C_n, kxC_n, kxR_4, kxU_1, kxU_2
 from .dynamic_buffer import DynamicBuffer
 
 __all__ = [
     "DynamicBuffer",
-    "U_1",
-    "U_2",
-    "U_4",
+    "B_1",
+    "C_1",
     "I_1",
     "I_2",
     "I_4",
     "R_4",
     "R_8",
-    "C_1",
-    "C_n",
-    "B_1",
+    "U_1",
+    "U_2",
+    "U_4",
     "B_n",
+    "C_n",
+    "kxC_n",
+    "kxR_4",
+    "kxU_1",
+    "kxU_2",
 ]

pystdf4/Core/data_base.py

@@ -1,6 +1,6 @@
 from abc import ABC, abstractmethod
 from struct import Struct
-from typing import Any, ClassVar, Generic, Sequence, TypeVar
+from typing import Any, ClassVar, Generic, Optional, Sequence, Type, TypeVar
 
 from pystdf4.Core.dynamic_buffer import DynamicBuffer
 from pystdf4.Core.mixins import CacheMixin
@@ -19,7 +19,7 @@
 class FieldBase(ABC, Generic[pyT]):
     @classmethod
     @abstractmethod
-    def pack_into(cls, buffer: DynamicBuffer, value: pyT):
+    def pack_into(cls, buffer: DynamicBuffer, value: pyT, size: int = 0):
         raise NotImplementedError()
 
     @classmethod
@@ -42,8 +42,8 @@ class ImmediateField(FieldBase[pyT]):
     field_size: ClassVar[int] = 0
 
     @classmethod
-    def pack_into(cls, buffer: DynamicBuffer, value: pyT):
-        data = cls._normalize_value(value)
+    def pack_into(cls, buffer: DynamicBuffer, value: pyT, size: int = 0):
+        data = cls._normalize_value(value, size)
         field_size = cls.field_size if cls.field_size else len(data)
         buffer._ensure_capacity(field_size)
         buffer._mv[buffer.offset : buffer.offset + field_size] = data
@@ -54,9 +54,13 @@ def pack_into(cls, buffer: DynamicBuffer, value: pyT):
     def _pascal_bytes(value: bytes) -> bytes:
         return bytes((len(value),)) + value
 
+    @staticmethod
+    def _left_justified_bytes(value: bytes, size: int, fill_byte: bytes) -> bytes:
+        return value.ljust(size, fill_byte)
+
     @staticmethod
     @abstractmethod
-    def _normalize_value(value: Any) -> bytes:
+    def _normalize_value(value: Any, size: int = 0) -> bytes:
         raise NotImplementedError()
 
 
@@ -72,7 +76,7 @@ def __init_subclass__(cls) -> None:
         cls.field_size = Struct(f"{cls.endian}{cls.num_elements}{cls.struct_format}").size
 
     @classmethod
-    def pack_into(cls, buffer: DynamicBuffer, value: pyT):
+    def pack_into(cls, buffer: DynamicBuffer, value: pyT, size: int = 0):
         """Reserve space, cache the value, and advance the buffer offset."""
         buffer._ensure_capacity(cls.field_size)
         cls._enqueue_value(value, buffer.offset, cls.field_size)
@@ -89,3 +93,17 @@ def _serialize_sequence(cls, sequence: Sequence[pyT]) -> bytes:
         count = len(cls.buffer_offsets) * cls.num_elements
         packer = Struct(f"{cls.endian}{count}{cls.struct_format}")
         return packer.pack(*sequence)
+
+
+class ArrayField:
+    element_type: ClassVar[Type[FieldBase]]
+
+    @classmethod
+    def pack_into(cls, buffer: DynamicBuffer, value: Optional[Sequence[pyT]], size: int):
+        if value is None:
+            return
+        if len(value) > size:
+            value = value[:size]
+
+        for element in value:
+            cls.element_type.pack_into(buffer, element)

pystdf4/Core/data_type.py

@@ -1,4 +1,4 @@
-from pystdf4.Core.data_base import DeferredField, ImmediateField
+from pystdf4.Core.data_base import ArrayField, DeferredField, ImmediateField
 
 # ============================================================
 # Numeric Field Implementations
@@ -42,29 +42,77 @@ class R_8(DeferredField[float]):
 # ============================================================
 
 
-class C_1(ImmediateField):
+class C_1(ImmediateField[str]):
     field_size = 1
 
     @staticmethod
-    def _normalize_value(value: str) -> bytes:
+    def _normalize_value(value: str, size: int = 0) -> bytes:
         return value.encode("ascii")
 
 
-class B_1(ImmediateField):
+class C_12(ImmediateField[str]):
+    field_size = 12
+
+    @staticmethod
+    def _normalize_value(value: str, size: int = 0) -> bytes:
+        return value.encode("ascii")
+
+
+class B_1(ImmediateField[bytes]):
     field_size = 1
 
     @staticmethod
-    def _normalize_value(value: bytes) -> bytes:
+    def _normalize_value(value: bytes, size: int = 0) -> bytes:
+        return value
+
+
+class B_6(ImmediateField[bytes]):
+    field_size = 6
+
+    @staticmethod
+    def _normalize_value(value: bytes, size: int = 0) -> bytes:
         return value
 
 
-class C_n(ImmediateField):
+class C_n(ImmediateField[str]):
     @staticmethod
-    def _normalize_value(value: str) -> bytes:
+    def _normalize_value(value: str, size: int = 0) -> bytes:
         return ImmediateField._pascal_bytes(value.encode("ascii"))
 
 
-class B_n(ImmediateField):
+class B_n(ImmediateField[bytes]):
     @staticmethod
-    def _normalize_value(value: bytes) -> bytes:
+    def _normalize_value(value: bytes, size: int = 0) -> bytes:
         return ImmediateField._pascal_bytes(value)
+
+
+class C_f(ImmediateField[str]):
+    @staticmethod
+    def _normalize_value(value: bytes, size: int = 0) -> bytes:
+        return ImmediateField._left_justified_bytes(value, size, b" ")
+
+
+# ============================================================
+# Array Field Implementations
+# ============================================================
+
+
+class kxU_1(ArrayField):
+    element_type = U_1
+
+
+class kxU_2(ArrayField):
+    element_type = U_2
+
+
+class kxC_n(ArrayField):
+    element_type = C_n
+
+
+class kxR_4(ArrayField):
+    element_type = R_4
+
+
+# TODO: Implement N_1 field!
+# class kxN_1(ArrayField):
+#     element_type = N_1

pystdf4/IO/__init__.py

@@ -1,4 +1,4 @@
 from .stdf4reader import Stdf4Reader
-from .stfd4write import Stfd4Writer
+from .stdf4writer import Stdf4Writer
 
-__all__ = ["Stdf4Reader", "Stfd4Writer"]
+__all__ = ["Stdf4Reader", "Stdf4Writer"]

pystdf4/IO/base.py

@@ -1,48 +0,0 @@
-from pathlib import Path
-from struct import Struct
-
-from pystdf4.Core import B_1, C_1, I_1, I_2, I_4, R_4, R_8, U_1, U_2, U_4, B_n, C_n
-from pystdf4.Core.dynamic_buffer import DynamicBuffer
-
-
-class StdfIOBase:
-    __slots__ = (
-        "file_path",
-        "buffer",
-        "B_1",
-        "C_1",
-        "I_1",
-        "I_2",
-        "I_4",
-        "R_4",
-        "R_8",
-        "U_1",
-        "U_2",
-        "U_4",
-        "B_n",
-        "C_n",
-        "KxC_n",
-        "KxR_4",
-        "KxU_1",
-        "KxU_2",
-        "header_packer",
-    )
-
-    def __init__(self, file_path: str):
-        self.buffer = DynamicBuffer()
-        self.file_path = Path(file_path)
-
-        self.B_1 = B_1()
-        self.C_1 = C_1()
-        self.I_1 = I_1()
-        self.I_2 = I_2()
-        self.I_4 = I_4()
-        self.R_4 = R_4()
-        self.R_8 = R_8()
-        self.U_1 = U_1()
-        self.U_2 = U_2()
-        self.U_4 = U_4()
-        self.B_n = B_n()
-        self.C_n = C_n()
-
-        self.header_packer: Struct = Struct("<HBB")

pystdf4/IO/stdf4reader.py

@@ -1,6 +1,6 @@
-from .base import StdfIOBase
+from pathlib import Path
 
 
-class Stdf4Reader(StdfIOBase):
+class Stdf4Reader:
     def __init__(self, file_path: str):
-        super().__init__(file_path)
+        self.file_path = Path(file_path)

pystdf4/IO/stdf4writer.py

@@ -0,0 +1,34 @@
+from pathlib import Path
+
+from pystdf4.Records.packer import StdfPacker
+
+
+class Stdf4Writer(StdfPacker):
+    # region Magic Methods
+    def __init__(self, file_path: str):
+        super(StdfPacker, self).__init__()
+        self.file_path = Path(file_path)
+
+    def __enter__(self) -> "Stdf4Writer":
+        if not self.file_path.exists():
+            self.file_path.touch()
+        return self
+
+    def __exit__(self, exc_type: type, exc_val: Exception, exc_tb: object) -> None:
+        self._write_buffers()
+        self.file_path.write_bytes(self.buffer.to_bytes())
+
+    # endregion
+
+    def _write_buffers(self):
+        for scalar_field in (
+            self.U_1,
+            self.U_2,
+            self.U_4,
+            self.I_1,
+            self.I_2,
+            self.I_4,
+            self.R_4,
+            self.R_8,
+        ):
+            scalar_field.flush_cache_to_buffer(self.buffer)