[feature] Bitfield record class

opendis/DataInputStream.py

@@ -61,3 +61,6 @@ def read_int(self) -> int32:
 
     def read_unsigned_int(self) -> uint32:
         return struct.unpack('>I', self.stream.read(4))[0]
+
+    def read_bytes(self, n: int) -> bytes:
+        return self.stream.read(n)

opendis/DataOutputStream.py

@@ -49,3 +49,6 @@ def write_int(self, val: int) -> None:
     def write_unsigned_int(self, val: int) -> None:
         self.stream.write(struct.pack('>I', val))
 
+    def write_bytes(self, val: bytes) -> None:
+        self.stream.write(val)
+

opendis/dis7.py

@@ -6,8 +6,6 @@
     enum8,
     enum16,
     enum32,
-    int8,
-    int16,
     int32,
     uint8,
     uint16,
@@ -19,6 +17,7 @@
     struct16,
     struct32,
 )
+from .record import SpreadSpectrum
 
 
 class DataQueryDatumSpecification:
@@ -2025,11 +2024,11 @@ class ModulationType:
     """
 
     def __init__(self,
-                 spreadSpectrum: struct16 = 0,  # See RPR Enumerations
+                 spreadSpectrum: SpreadSpectrum | None = None,  # See RPR Enumerations
                  majorModulation: enum16 = 0,  # [UID 155]
-                 detail: enum16  =0,  # [UID 156-162]
-                 radioSystem: enum16  =0):  # [UID 163]
-        self.spreadSpectrum = spreadSpectrum
+                 detail: enum16 = 0,  # [UID 156-162]
+                 radioSystem: enum16 = 0):  # [UID 163]
+        self.spreadSpectrum = spreadSpectrum or SpreadSpectrum()
         """This field shall indicate the spread spectrum technique or combination of spread spectrum techniques in use. Bit field. 0=freq hopping, 1=psuedo noise, time hopping=2, reamining bits unused"""
         self.majorModulation = majorModulation
         """the major classification of the modulation type."""
@@ -2040,14 +2039,14 @@ def __init__(self,
 
     def serialize(self, outputStream):
         """serialize the class"""
-        outputStream.write_unsigned_short(self.spreadSpectrum)
+        self.spreadSpectrum.serialize(outputStream)
         outputStream.write_unsigned_short(self.majorModulation)
         outputStream.write_unsigned_short(self.detail)
         outputStream.write_unsigned_short(self.radioSystem)
 
     def parse(self, inputStream):
         """Parse a message. This may recursively call embedded objects."""
-        self.spreadSpectrum = inputStream.read_unsigned_short()
+        self.spreadSpectrum.parse(inputStream)
         self.majorModulation = inputStream.read_unsigned_short()
         self.detail = inputStream.read_unsigned_short()
         self.radioSystem = inputStream.read_unsigned_short()
@@ -5441,7 +5440,7 @@ def __init__(self,
                  antennaPatternList=None):
         super(TransmitterPdu, self).__init__()
         self.radioReferenceID = radioReferenceID or EntityID()
-        """ID of the entitythat is the source of the communication"""
+        """ID of the entity that is the source of the communication"""
         self.radioNumber = radioNumber
         """particular radio within an entity"""
         self.radioEntityType = radioEntityType or EntityType()  # TODO: validation
@@ -5455,7 +5454,6 @@ def __init__(self,
         self.frequency = frequency
         self.transmitFrequencyBandwidth = transmitFrequencyBandwidth
         self.power = power
-        """transmission power"""
         self.modulationType = modulationType or ModulationType()
         self.cryptoSystem = cryptoSystem
         self.cryptoKeyId = cryptoKeyId
@@ -5464,9 +5462,7 @@ def __init__(self,
         self.padding2 = 0
         self.padding3 = 0
         self.modulationParametersList = modulationParametersList or []
-        """variable length list of modulation parameters"""
         self.antennaPatternList = antennaPatternList or []
-        """variable length list of antenna pattern records"""
         # TODO: zero or more Variable Transmitter Parameters records (see 6.2.95)
 
     @property
@@ -7463,9 +7459,11 @@ def parse(self, inputStream):
         self.encodingScheme = inputStream.read_unsigned_short()
         self.tdlType = inputStream.read_unsigned_short()
         self.sampleRate = inputStream.read_unsigned_int()
-        self.dataLength = inputStream.read_unsigned_short()
+        dataLength = inputStream.read_unsigned_short()
+        # TODO: Make validation optional
+        assert dataLength % 8 == 0
         self.samples = inputStream.read_unsigned_short()
-        for idx in range(0, self.dataLength // 8):
+        for idx in range(0, dataLength // 8):
             element = inputStream.read_unsigned_byte()
             self.data.append(element)
 

opendis/record.py

@@ -0,0 +1,143 @@
+from collections.abc import Sequence
+from ctypes import c_uint8, c_uint16, BigEndianStructure
+
+from .types import (
+    bf_enum,
+    bf_uint,
+)
+from .DataInputStream import DataInputStream
+from .DataOutputStream import DataOutputStream
+
+
+def bitfield(
+        name: str,
+        bytesize: int,
+        fields: Sequence[tuple],
+    ):
+    """Factory function for bitfield structs, which are subclasses of
+    ctypes.Structure.
+    These are used in records that require them to unpack non-octet-sized fields.
+
+    Arguments
+    ---------
+    name: str
+        The name of the bitfield struct.
+    bytesize: int
+        The number of bytes required by the bitfield, including padding.
+    fields: Sequence[tuple[str, ctypes._CData] | tuple[str, ctypes._CData, int]]
+        A sequence of field descriptions. See ctypes.Structure documentation for details.
+    """
+    assert bytesize > 0, "Cannot create bitfield with zero bytes"
+
+    class Bitfield(BigEndianStructure):
+        _fields_ = fields
+
+        @staticmethod
+        def marshalledSize() -> int:
+            return bytesize
+
+        def serialize(self, outputStream: DataOutputStream) -> None:
+            outputStream.write_bytes(bytes(self))
+
+        @classmethod
+        def parse(cls, inputStream: DataInputStream) -> "Bitfield":
+            return cls.from_buffer_copy(inputStream.read_bytes(bytesize))
+    Bitfield.__name__ = name
+    return Bitfield
+
+
+class NetId:
+    """Annex C, Table C.5
+
+    Represents an Operational Net in the format of NXX.XYY, where:
+    N = Mode
+    XXX = Net Number
+    YY = Frequency Table
+    """
+    _struct = bitfield("NetId", 2, [
+        ("netNumber", c_uint16, 10),
+        ("frequencyTable", c_uint8, 2),
+        ("mode", c_uint8, 2),
+        ("padding", c_uint8, 2)
+    ])
+
+    def __init__(self,
+                 netNumber: bf_uint = 0,
+                 frequencyTable: bf_enum = 0,  # [UID 299]
+                 mode: bf_enum = 0,  # [UID 298]
+                 padding: bf_uint = 0):
+        # Net number ranging from 0 to 999 decimal
+        self.netNumber = netNumber
+        self.frequencyTable = frequencyTable
+        self.mode = mode
+        self.padding = padding
+
+    def marshalledSize(self) -> int:
+        return self._struct.marshalledSize()
+
+    def serialize(self, outputStream: DataOutputStream) -> None:
+        self._struct(
+            self.netNumber,
+            self.frequencyTable,
+            self.mode,
+            self.padding
+        ).serialize(outputStream)
+
+    def parse(self, inputStream: DataInputStream) -> None:
+        record_bitfield = self._struct.parse(inputStream)
+        self.netNumber = record_bitfield.netNumber
+        self.frequencyTable = record_bitfield.frequencyTable
+        self.mode = record_bitfield.mode
+        self.padding = record_bitfield.padding
+
+
+class SpreadSpectrum:
+    """6.2.59 Modulation Type Record, Table 90
+
+    Modulation used for radio transmission is characterized in a generic
+    fashion by the Spread Spectrum, Major Modulation, and Detail fields.
+
+    Each independent type of spread spectrum technique shall be represented by
+    a single element of this array.
+    If a particular spread spectrum technique is in use, the corresponding array
+    element shall be set to one; otherwise it shall be set to zero.
+    All unused array elements shall be set to zero.
+
+    In Python, the presence or absence of each technique is indicated by a bool.
+    """
+    _struct = bitfield("SpreadSpectrum", 2, [
+        ("frequencyHopping", c_uint8, 1),
+        ("pseudoNoise", c_uint8, 1),
+        ("timeHopping", c_uint8, 1),
+        ("padding", c_uint16, 13)
+    ])
+
+    def __init__(self,
+                 frequencyHopping: bool = False,
+                 pseudoNoise: bool = False,
+                 timeHopping: bool = False,
+                 padding: bf_uint = 0):
+        self.frequencyHopping = frequencyHopping
+        self.pseudoNoise = pseudoNoise
+        self.timeHopping = timeHopping
+        self.padding = padding
+
+    def marshalledSize(self) -> int:
+        return self._struct.marshalledSize()
+
+    def serialize(self, outputStream: DataOutputStream) -> None:
+        # Bitfield expects int input
+        self._struct(
+            int(self.frequencyHopping),
+            int(self.pseudoNoise),
+            int(self.timeHopping),
+            self.padding
+        ).serialize(outputStream)
+
+    def parse(self, inputStream: DataInputStream) -> None:
+        # Pass bool to __init__ instead of int
+        record_bitfield = self._struct.parse(inputStream)
+        self.frequencyHopping = bool(record_bitfield.frequencyHopping)
+        self.pseudoNoise = bool(record_bitfield.pseudoNoise)
+        self.timeHopping = bool(record_bitfield.timeHopping)
+

opendis/types.py

@@ -27,3 +27,8 @@
 struct32 = bytes
 char8 = str
 char16 = str
+
+# Non-octet-size types for bitfields
+bf_enum = int
+bf_int = int
+bf_uint = int