Changeset a8f6b97 in ammosreader for ammosreader/PDW.py

Timestamp:

09/03/23 17:01:14 (21 months ago)

Author:

Enrico Schwass <ennoausberlin@…>

Branches:

guix

Children:

be36a02

Parents:

2b95c18

Message:

bitstring library added as dependency

lots of properties added, float values removed, scaling and signed integer extension fixed

File:

• ammosreader/PDW.py (modified) (15 diffs)

ammosreader/PDW.py

@@ -4 +4 @@
 import math
 import numpy as np
+from bitstring import BitArray
 from ammosreader import logger
 
@@ -13 +14 @@
     """
 
+    @classmethod
+    def polarities(cls):
+         return {0: 'Horizontal/Unknown', 1: 'Vertical', 2: 'Counter clockwise', 3: 'Clockwise'}
+
+    @classmethod
+    def modulations(cls):
+        return {0: 'Unknown', 1: 'Unmodulated', 2: 'FM', 3: 'LFM', 4: 'PSK-2', 5: 'PSK-3', 6: 'PSK-4',
+                7: 'PSK-m', 8: 'NLFM', 9: 'SFM', 10: 'TFM', 11: 'Pulse too short'}
+    
     @classmethod
     def from_bytes(cls, byte_string):
@@ -36 +46 @@
         if nanoseconds >= unix_time:
             raise OverflowError("Timestamp invalid")
-        time_of_arrival = np.datetime64(nanoseconds, 'ns')
-
-        third_entry = bin(int.from_bytes(parts[1], byteorder='little'))
-        padding = 32-len(str(third_entry)[2:])
-        third_entry_bit_string = "0" * padding + str(third_entry)[2:]
-        pdw_format_identifier = int(third_entry_bit_string[0:6], 2)
-        center_frequency = int(third_entry_bit_string[5:32], 2)
-
-        fourth_entry = bin(int.from_bytes(parts[2], byteorder='little'))
-        padding = 32-len(str(fourth_entry)[2:])
-        fourth_entry_bit_string = "0" * padding + str(fourth_entry)[2:]
-        is_valid = bool(int(fourth_entry_bit_string[0]))
-        is_pulse = bool(int(fourth_entry_bit_string[1]))
-        level_unit = bool(int(fourth_entry_bit_string[2]))
-        signal_start_missing = bool(int(fourth_entry_bit_string[3]))
-        signal_end_missing = bool(int(fourth_entry_bit_string[4]))
-        pulse_width = int(fourth_entry_bit_string[7:33], 2)
-
-        fifth_entry = bin(int.from_bytes(parts[3], byteorder='little'))
-        padding = 32-len(str(fifth_entry)[2:])
-        fifth_entry_bit_string = "0" * padding + str(fifth_entry)[2:]
-        frequency_shift_or_bandwidth = int(fifth_entry_bit_string[0:20], 2)
-        # FIXME: You have to scale me to the range from -200.0 to 200.0 in 0.1 steps
-        pulse_level_or_pulse_field_strength = math.ceil(int(fifth_entry_bit_string[20:32], 2)) / 10
-
-        sixth_entry = bin(int.from_bytes(parts[4], byteorder='little'))
-        padding = 32-len(str(sixth_entry)[2:])
-        sixth_entry_bit_string = "0" * padding + str(sixth_entry)[2:]
-        region_of_interest = bool(int(sixth_entry_bit_string[0]))
-        # FIXME: You have to scale me to a range from 0.0 to 6.2 in steps of 0.1 - 6.3 means unknown
-        azimuth_confidence = math.ceil(int(sixth_entry_bit_string[1:7], 2)) / 10
-        modulations = {0: 'Unknown', 1: 'Unmodulated', 2: 'FM', 3: 'LFM', 4: 'PSK-2', 5: 'PSK-3', 6: 'PSK-4',
-                       7: 'PSK-m', 8: 'NLFM', 9: 'SFM', 10: 'TFM', 11: 'Pulse too short'}
-        modulation = modulations.get(int(sixth_entry_bit_string[7:12], 2), 0)
-        sector = int(sixth_entry_bit_string[28:32], 2)
-
-        seventh_entry = bin(int.from_bytes(parts[5], byteorder='little'))
-        padding = 32-len(str(seventh_entry)[2:])
-        seventh_entry_bit_string = "0" * padding + str(seventh_entry)[2:]
-        polarities = {0: 'Horizontal/Unknown', 1: 'Vertical', 2: 'Counter clockwise', 3: 'Clockwise'}
-        polarity = polarities[int(seventh_entry_bit_string[0:2], 2)]
-        df_quality = int(seventh_entry_bit_string[2:9], 2)
-        # FIXME: You have to scale me from -90 to 90 in 0.1 degree steps
-        elevation = int(seventh_entry_bit_string[9:20], 2) / 10
-        # FIXME: You have to check me for a range from 0.0 to 359.9 in steps of 0.1
-        azimuth = 0.1 * (int(seventh_entry_bit_string[20:32], 2))
-
-        eighth_entry = bin(int.from_bytes(parts[5], byteorder='little'))
-        padding = 32-len(str(eighth_entry)[2:])
-        eighth_entry_bit_string = "0" * padding + str(eighth_entry)[2:]
-        channel = int(eighth_entry_bit_string[0:4], 2)
+        time_of_arrival = nanoseconds
+
+        third_word = bin(int.from_bytes(parts[1], byteorder='little'))
+        padding = 32-len(str(third_word)[2:])
+        third_word_bit_string = "0" * padding + str(third_word)[2:]
+        pdw_format_identifier = int(third_word_bit_string[0:6], 2)
+        center_frequency = int(third_word_bit_string[5:32], 2) # Value of zero means unknown
+        
+        fourth_word = bin(int.from_bytes(parts[2], byteorder='little'))
+        padding = 32-len(str(fourth_word)[2:])
+        fourth_word_bit_string = "0" * padding + str(fourth_word)[2:]
+        is_valid = bool(int(fourth_word_bit_string[0]))
+        is_pulse = bool(int(fourth_word_bit_string[1]))
+        level_unit = int(fourth_word_bit_string[2])
+        signal_start_missing = bool(int(fourth_word_bit_string[3]))
+        signal_end_missing = bool(int(fourth_word_bit_string[4]))
+        pulse_width = int(fourth_word_bit_string[7:33], 2)
+
+        fifth_word = bin(int.from_bytes(parts[3], byteorder='little'))
+        padding = 32-len(str(fifth_word)[2:])
+        fifth_word_bit_string = "0" * padding + str(fifth_word)[2:]
+        frequency_shift_or_bandwidth = int(fifth_word_bit_string[0:20], 2)
+        pulse_level_or_pulse_field_strength = BitArray(fifth_word_bit_string[20:32]).int
+
+        sixth_word = bin(int.from_bytes(parts[4], byteorder='little'))
+        padding = 32-len(str(sixth_word)[2:])
+        sixth_word_bit_string = "0" * padding + str(sixth_word)[2:]
+        region_of_interest = bool(int(sixth_word_bit_string[0]))
+        azimuth_confidence = int(sixth_word_bit_string[1:7], 2)
+        modulation = int(sixth_word_bit_string[7:12], 2)
+        sector = int(sixth_word_bit_string[28:32], 2)
+
+        seventh_word = bin(int.from_bytes(parts[5], byteorder='little'))
+        padding = 32-len(str(seventh_word)[2:])
+        seventh_word_bit_string = "0" * padding + str(seventh_word)[2:]
+        polarity = int(seventh_word_bit_string[0:2], 2)
+        df_quality = int(seventh_word_bit_string[2:9], 2)
+        elevation = BitArray(bin=seventh_word_bit_string[9:20]).int
+        azimuth = int(seventh_word_bit_string[20:32], 2)
+
+        eighth_word = bin(int.from_bytes(parts[5], byteorder='little'))
+        padding = 32-len(str(eighth_word)[2:])
+        eighth_word_bit_string = "0" * padding + str(eighth_word)[2:]
+        channel = int(eighth_word_bit_string[0:4], 2)
 
         return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit,
@@ -110 +113 @@
         :param is_pulse: flag to mark if pdw data body contains a pulse or a continuous wave signal
         :type is_pulse: Boolean
-        :param level_unit: 0 means dBµV - 1 means dBµV/m
+        :param level_unit: 1 means dBµV - 0 means dBµV/m
         :type level_unit: Integer
         :param signal_start_missing: signal started before time of arrival
@@ -121 +124 @@
         :type frequency_shift_or_bandwidth: Integer
         :param pulse_level_or_pulse_field_strength: Pulse level or Pulse Field Strength depending on level_unit \
-         (-200.0...200.0) in 0.1 steps / minus 204.8 means no valid level detected
-        :type pulse_level_or_pulse_field_strength: Float
+         (-200.0...200.0) in tenth degrees / minus 2048 means no valid level detected
+        :type pulse_level_or_pulse_field_strength: Integer
         :param region_of_interest: Marks if signal is from region of interest
         :type region_of_interest: Boolean
-        :param azimuth_confidence: degree in steps of 0.1 (0.0-6.2) / 6.3 means confidence unknown
-        :type azimuth_confidence: Float
+        :param azimuth_confidence: degree in tenth steps of (0-62) / 63 means confidence unknown
+        :type azimuth_confidence: Integer
         :param modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.)
         :type modulation: String
@@ -135 +138 @@
         :param df_quality: Direction finding quality in percent (0-100) - Zero means unknown
         :type df_quality: Integer
-        :param elevation: elevation of incoming signal (from -90 to 90 degree) in steps of 0.1 degree \
-        minus 102.4 means unknown
-        :type elevation: Float
-        :param azimuth: azimuth of incoming signal (from 0 to 359.9 degree) in steps of 0.1 degree \
-        plus 409.5 means unknown
-        :type azimuth: Float
+        :param elevation: elevation of incoming signal (from -90 to 90 degree) in steps tenths degree \
+        minus 1024 means unknown
+        :type elevation: Integer
+        :param azimuth: azimuth of incoming signal (from 0 to 3599 tenth degree) plus 4095 means unknown
+        :type azimuth: Integer
         :param channel: detecting channel (0-16) - Zero means unknown
         :type channel: Integer
@@ -146 +148 @@
         :rtype: PDW
         """
-        self.time_of_arrival = time_of_arrival #
-        self.pdw_format_identifier = pdw_format_identifier
-        self.center_frequency = center_frequency #
+        self.__time_of_arrival = time_of_arrival #
+        self.__pdw_format_identifier = pdw_format_identifier
+        self.__center_frequency = center_frequency #
         self.__is_valid = is_valid #
-        self.is_pulse = is_pulse #
-        self.level_unit = level_unit #
-        self.signal_start_missing = signal_start_missing
-        self.signal_end_missing = signal_end_missing
-        self.pulse_width = pulse_width #
-        self.frequency_shift_or_bandwidth = frequency_shift_or_bandwidth #
-        self.pulse_level_or_pulse_field_strength = pulse_level_or_pulse_field_strength #
-        self.region_of_interest = region_of_interest
-        self.azimuth_confidence = azimuth_confidence
-        self.modulation = modulation #
-        self.sector = sector
-        self.polarity = polarity
-        self.df_quality = df_quality #
-        self.elevation = elevation #
-        self.azimuth = azimuth
-        self.channel = channel #
+        self.__is_pulse = is_pulse #
+        self.__level_unit = level_unit #
+        self.__signal_start_missing = signal_start_missing
+        self.__signal_end_missing = signal_end_missing
+        self.__pulse_width = pulse_width #
+        self.__frequency_shift_or_bandwidth = frequency_shift_or_bandwidth #
+        self.__pulse_level_or_pulse_field_strength = pulse_level_or_pulse_field_strength #
+        self.__region_of_interest = region_of_interest
+        self.__azimuth_confidence = azimuth_confidence
+        self.__modulation = modulation #
+        self.__sector = sector
+        self.__polarity = polarity
+        self.__df_quality = df_quality #
+        self.__elevation = elevation #
+        self.__azimuth = azimuth
+        self.__channel = channel #
 
     def __str__(self):
@@ -177 +179 @@
                   "Center frequency: " + str(self.center_frequency) + " KHz\n")
 
-        if self.__is_valid:
+        if self.is_valid:
             output += "Signal: Valid\n"
         else:
@@ -187 +189 @@
             output += "Signal type: Continuous wave\n"
 
-        if self.level_unit == 1:
+        # FIXME: use fstrings or another better performance string concat
+        if self.level_unit == "dbµV":
             output += "Pulse level: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV\n"
         else:
-            output += "Pulse field strength: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV/meter\n"
+            output += "Pulse field strength: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV/m\n"
 
         output += ("Pulse width: " + str(self.pulse_width) + " nanoseconds\n" +
@@ -200 +203 @@
             output += "Region of interest: No\n"
 
-        if self.azimuth_confidence == 6.3:
+        if self.azimuth_confidence == None:
             output += "Azimuth confidence: Invalid\n"
         else:
@@ -207 +210 @@
         output += "Modulation: " + str(self.modulation) + "\n"
 
-        if self.sector == 0:
+        if self.sector == None:
             output += "Sector: Unknown\n"
         else:
@@ -216 +219 @@
         output += "DF quality: " + str(self.df_quality) + " %\n"
 
-        if self.elevation == 102.4:
+        if self.elevation == None:
             output += "Elevation: Unknown\n"
         else:
             output += "Elevation: " + str(self.elevation) + " degree\n"
 
-        if self.azimuth == 409.5:
+        if self.azimuth == None:
             output += "Azimuth: Unknown\n"
         else:
@@ -230 +233 @@
         return output
 
+    @property
+    def time_of_arrival(self):
+        return np.datetime64(self.__time_of_arrival, 'ns')
+    
+    @property
+    def center_frequency(self):
+        if self.__center_frequency == 0:
+            return None
+        return self.__center_frequency
+
+    @property
     def is_valid(self):
         return self.__is_valid
 
+    @property
+    def is_invalid(self):
+        return not self.__is_valid
+
+    @property
+    def is_pulse(self):
+        return self.__is_pulse
+
+    @property
+    def is_cw(self):
+        return not self.__is_pulse
+
+    @property
+    def level_unit(self):
+        if self.__level_unit == 1:
+            return "dbµV"
+        else:
+            return "dbµV/m"
+
+    @property
+    def signal_start_missing(self):
+        return self.__signal_start_missing
+
+    @property
+    def signal_end_missing(self):
+        return self.__signal_end_missing
+
+    @property
+    def pulse_width(self):
+        if self.__pulse_width == 0:
+            return None
+        return self.__pulse_width
+
+    @property
+    def frequency_shift_or_bandwidth(self):
+        if self.__frequency_shift_or_bandwidth == 1048575:
+            return None
+        return self.__frequency_shift_or_bandwidth
+
+    @property
+    def pulse_level_or_pulse_field_strength(self):
+        if self.__pulse_level_or_pulse_field_strength == -2048:
+            return None
+        return (self.__pulse_level_or_field_strength / 10) 
+
+    @property
+    def azimuth_confidence(self):
+        if azimuth_confidence == 63:
+            return None
+        return (azimuth_confidence / 10)
+
+    @property
+    def modulation(self):
+        return PDW.modulations().get(self.__modulation, PDW.modulations()[0])
+    
+    @property
+    def region_of_interest(self):
+        return self.__region_of_interest
+
+    @property
+    def sector(self):
+        if self.__sector == 0:
+            return None
+        return self.__sector
+
+    @property
+    def polarity(self):
+        return PDW.polarities().get(self.__polarity, PDW.polarities()[0])
+
+    @property
+    def elevation(self):
+        if self.__elevation == -1024:
+            return None
+        return self.__elevation / 10
+
+    @property
+    def azimuth(self):
+        if self.__azimuth == 4095:
+            return None
+        return self.__azimuth / 10
+
+    @property
+    def channel(self):
+        return self.__channel
+    
     def to_datacrunch_json(self):
         import uuid
@@ -238 +337 @@
                 'FORMAT': self.pdw_format_identifier,
                 'CENTER_FREQUENCY': self.center_frequency,
-                'VALID_FLAG': self.__is_valid,
+                'VALID_FLAG': self.is_valid,
                 'PULSE_FLAG': self.is_pulse,
                 'LU_FLAG': self.level_unit,
@@ -259 +358 @@
 
     def to_json(self):
-        return {'time of arrival': self.time_of_arrival,
+        return {'TIME OF ARRIVAL': self.time_of_arrival,
                 'FORMATIDENTIFIER': self.pdw_format_identifier,
-                'center frequency': self.center_frequency,
-                'VALID': self.__is_valid,
+                'CENTER FREQUENCY': self.center_frequency,
+                'VALID': self.is_valid,
                 'PULSE': self.is_pulse,
                 'PULSELEVEL': self.pulse_level_or_pulse_field_strength,