Index: ammosreader/AmmosGlobalFrameHeader.py =================================================================== --- ammosreader/AmmosGlobalFrameHeader.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ ammosreader/AmmosGlobalFrameHeader.py (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -1,8 +1,5 @@ """I provide an AMMOS global frame header.""" import struct -import logging - -logging.basicConfig(filename='ammos.log', level=logging.DEBUG) - +from ammosreader import logger class AmmosGlobalFrameHeader: @@ -21,4 +18,5 @@ magic_word = elements[0].hex() + logger.debug("Header created") if magic_word != cls.MAGIC_WORD: return None @@ -40,5 +38,5 @@ """I return a new instance of myself initialized with above parameters.""" if magic_word != type(self).MAGIC_WORD: - logging.error("Wrong magic word found") + logger.error("Wrong magic word found") self.magic_word = magic_word else: Index: ammosreader/PDW.py =================================================================== --- ammosreader/PDW.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ ammosreader/PDW.py (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -4,5 +4,5 @@ import math import numpy as np - +from ammosreader import logger class PDW(): @@ -24,8 +24,16 @@ :rtype: PDW """ - assert(len(byte_string) == 32) + + logger.info("from bytes") + if (len(byte_string) != 32): + logger.error("Byte count invalid") + raise TypeError("Byte count invalid") parts = struct.unpack('= unix_time: + raise OverflowError("Timestamp invalid") time_of_arrival = np.datetime64(nanoseconds, 'ns') @@ -61,5 +69,5 @@ 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[int(sixth_entry_bit_string[7:12], 2)] + modulation = modulations.get(int(sixth_entry_bit_string[7:12], 2), 0) sector = int(sixth_entry_bit_string[28:32], 2) Index: ammosreader/__init__.py =================================================================== --- ammosreader/__init__.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ ammosreader/__init__.py (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,29 @@ +import logging +import logging.config +import os +from pathlib import Path + +# change the default log path to /var/log/ammos/ammos.log when system configuration created this dir +# with the appropriate rights + +log_dir = Path(os.environ.get('AMMOS_LOG_DIR', '/tmp/')) + +log_path = log_dir / 'ammos.log' + +if not log_path.exists(): +# print(log_path, "does not exist") + try: +# print("Trying to create logfile", str(log_path)) + log_path.touch() + except PermissionError: +# print("Logging to file disabled") + conf_file = Path(__file__).parent / 'ammos_logging.conf' + print("Conf file", conf_file) + logging.config.fileConfig(conf_file) +else: +# print("Logging to", str(log_path)) + logging.basicConfig(filename=str(log_path), encoding='utf-8', level=logging.DEBUG) + +logger = logging.getLogger(__name__) + +logger.warning("Text") Index: ammosreader/ammos_logging.conf =================================================================== --- ammosreader/ammos_logging.conf (revision dff8988960eb5307013e281ff749d9fa74557406) +++ ammosreader/ammos_logging.conf (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,21 @@ +[loggers] +keys=root + +[handlers] +keys=stream_handler + +[formatters] +keys=formatter + +[logger_root] +level=DEBUG +handlers=stream_handler + +[handler_stream_handler] +class=StreamHandler +level=DEBUG +formatter=formatter +args=(sys.stderr,) + +[formatter_formatter] +format=%(asctime)s %(name)-12s %(levelname)-8s %(message)s Index: doc/ammosreader/AbstractAmmosReader.html =================================================================== --- doc/ammosreader/AbstractAmmosReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AbstractAmmosReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,648 @@ + + + + + + +ammosreader.AbstractAmmosReader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AbstractAmmosReader

+
+
+

I provide a base class for specialized AmmosReaders.

+
+ +Expand source code + +
"""I provide a base class for specialized AmmosReaders."""
+import logging
+
+from abc import ABC, abstractmethod
+
+from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader
+from ammosreader.AmmosSingleFrame import AmmosSingleFrame
+from ammosreader.AmmosContainer import AmmosContainer
+from ammosreader.AmmosConstants import FrameType
+
+logging.basicConfig(filename='ammos.log', level=logging.DEBUG)
+
+
+class AbstractAmmosReader(ABC):
+    """I implement a base class for specialized AmmosReaders."""
+
+    def __init__(self, file_name):
+        """
+        I am the standard constructor for Ammos Readers.
+
+        Additional information about the file can be added as key/value pairs in tags
+
+        :param file_name: The file to read Ammos data from
+        :type file_name: str
+        """
+        self.__file_name = file_name
+        self.__ammos_file = open(self.file_name, "rb")
+        self.__container = AmmosContainer(self.file_name, [])
+        self.__tags = {}
+
+    @property
+    def file_name(self):
+        """I return the name of the original file."""
+        return self.__file_name
+
+    @property
+    def ammos_file(self):
+        """I return the file to read the data from."""
+        return self.__ammos_file
+
+    @property
+    def container(self):
+        """I return the container which stores the data read."""
+        return self.__container
+
+    @property
+    def tags(self):
+        """I return all the tags."""
+        return self.__tags
+
+    def add_tag(self, a_key, a_value):
+        """I add information to tags using a key/value pair."""
+        assert a_key not in self.__tags
+        self.__tags[a_key] = a_value
+
+    def rewind_to_start(self):
+        """I set the file pointer to the beginning of the file for the next operation."""
+        self.ammos_file.seek(0)
+
+    def read_all_frames_left(self):
+        """
+        I read all remaining frames into my container until end of file is reached.
+
+        :return: a container containing all frames read
+        :rtype: AmmosContainer
+        """
+        frames_read = 0
+        while True:
+            logging.info("Reading single frame %s ...", frames_read)
+            current_frame = self.read_next_single_frame()
+            if current_frame is not None:
+                frames_read += 1
+                self.container.add_frame(current_frame)
+            else:
+                logging.debug("Frame: %s incomplete", frames_read+1)
+                break
+        logging.info("%s frames read", len(self.container.global_frames))
+        return self.container
+
+    def read_next_global_frame_header(self):
+        """
+        I return the next global frame header read from current position in file.
+
+        :return: the next global frame header or None if incomplete
+        :rtype: AmmosGlobalFrameHeader
+        """
+        header_size = AmmosGlobalFrameHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+        logging.info("Reading next global frame header")
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame header", header_size)
+            return None
+
+        current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes)
+
+        if current_global_frame_header is None:
+            return None
+
+        # logging.info("Current global frame header %s", current_global_frame_header)
+        return current_global_frame_header
+
+    @abstractmethod
+    def read_next_global_frame_body(self, data_header_length):
+        """My descendents have to implement this."""
+        pass
+
+    def read_next_single_frame(self):
+        """
+        I read and return a single global frame.
+
+        :return: a single global frame
+        :rtype: AmmosSingleFrame
+        """
+        global_frame_header = self.read_next_global_frame_header()
+
+        if global_frame_header is None:
+            return None
+
+        if global_frame_header is None:
+            logging.debug("Global frame header missing")
+            return None
+
+        if global_frame_header.data_header_length is None:
+            logging.debug("Data header length empty")
+            return None
+
+        if global_frame_header.frame_type not in list(FrameType):
+            logging.info("Unknown frame type %s found", global_frame_header.frame_type)
+            return None
+
+        # FIXME: Refactor duplicate logic - This code stinks
+
+        global_frame_body = None
+
+        if_data_types = [FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT.value,
+                         FrameType.IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT.value,
+                         FrameType.IF_DATA_16BIT_REAL_REAL.value,
+                         FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED.value,
+                         FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED.value]
+
+        if global_frame_header.frame_type == FrameType.AUDIO_DATA.value:
+            logging.info("Audio Datastream found")
+            global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
+            if global_frame_body is None:
+                return None
+            return AmmosSingleFrame(global_frame_header, global_frame_body)
+
+        if global_frame_header.frame_type in if_data_types:
+            logging.info("IF Datastream found")
+            global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
+            if global_frame_body is None:
+                return None
+            return AmmosSingleFrame(global_frame_header, global_frame_body)
+
+        if global_frame_body is None:
+            return None
+
+        logging.info("Unsupported frame type %s found", global_frame_header.frame_type)
+        return None
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AbstractAmmosReader +(file_name) +
+
+

I implement a base class for specialized AmmosReaders.

+

I am the standard constructor for Ammos Readers.

+

Additional information about the file can be added as key/value pairs in tags

+

:param file_name: The file to read Ammos data from +:type file_name: str

+
+ +Expand source code + +
class AbstractAmmosReader(ABC):
+    """I implement a base class for specialized AmmosReaders."""
+
+    def __init__(self, file_name):
+        """
+        I am the standard constructor for Ammos Readers.
+
+        Additional information about the file can be added as key/value pairs in tags
+
+        :param file_name: The file to read Ammos data from
+        :type file_name: str
+        """
+        self.__file_name = file_name
+        self.__ammos_file = open(self.file_name, "rb")
+        self.__container = AmmosContainer(self.file_name, [])
+        self.__tags = {}
+
+    @property
+    def file_name(self):
+        """I return the name of the original file."""
+        return self.__file_name
+
+    @property
+    def ammos_file(self):
+        """I return the file to read the data from."""
+        return self.__ammos_file
+
+    @property
+    def container(self):
+        """I return the container which stores the data read."""
+        return self.__container
+
+    @property
+    def tags(self):
+        """I return all the tags."""
+        return self.__tags
+
+    def add_tag(self, a_key, a_value):
+        """I add information to tags using a key/value pair."""
+        assert a_key not in self.__tags
+        self.__tags[a_key] = a_value
+
+    def rewind_to_start(self):
+        """I set the file pointer to the beginning of the file for the next operation."""
+        self.ammos_file.seek(0)
+
+    def read_all_frames_left(self):
+        """
+        I read all remaining frames into my container until end of file is reached.
+
+        :return: a container containing all frames read
+        :rtype: AmmosContainer
+        """
+        frames_read = 0
+        while True:
+            logging.info("Reading single frame %s ...", frames_read)
+            current_frame = self.read_next_single_frame()
+            if current_frame is not None:
+                frames_read += 1
+                self.container.add_frame(current_frame)
+            else:
+                logging.debug("Frame: %s incomplete", frames_read+1)
+                break
+        logging.info("%s frames read", len(self.container.global_frames))
+        return self.container
+
+    def read_next_global_frame_header(self):
+        """
+        I return the next global frame header read from current position in file.
+
+        :return: the next global frame header or None if incomplete
+        :rtype: AmmosGlobalFrameHeader
+        """
+        header_size = AmmosGlobalFrameHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+        logging.info("Reading next global frame header")
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame header", header_size)
+            return None
+
+        current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes)
+
+        if current_global_frame_header is None:
+            return None
+
+        # logging.info("Current global frame header %s", current_global_frame_header)
+        return current_global_frame_header
+
+    @abstractmethod
+    def read_next_global_frame_body(self, data_header_length):
+        """My descendents have to implement this."""
+        pass
+
+    def read_next_single_frame(self):
+        """
+        I read and return a single global frame.
+
+        :return: a single global frame
+        :rtype: AmmosSingleFrame
+        """
+        global_frame_header = self.read_next_global_frame_header()
+
+        if global_frame_header is None:
+            return None
+
+        if global_frame_header is None:
+            logging.debug("Global frame header missing")
+            return None
+
+        if global_frame_header.data_header_length is None:
+            logging.debug("Data header length empty")
+            return None
+
+        if global_frame_header.frame_type not in list(FrameType):
+            logging.info("Unknown frame type %s found", global_frame_header.frame_type)
+            return None
+
+        # FIXME: Refactor duplicate logic - This code stinks
+
+        global_frame_body = None
+
+        if_data_types = [FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT.value,
+                         FrameType.IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT.value,
+                         FrameType.IF_DATA_16BIT_REAL_REAL.value,
+                         FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED.value,
+                         FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED.value]
+
+        if global_frame_header.frame_type == FrameType.AUDIO_DATA.value:
+            logging.info("Audio Datastream found")
+            global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
+            if global_frame_body is None:
+                return None
+            return AmmosSingleFrame(global_frame_header, global_frame_body)
+
+        if global_frame_header.frame_type in if_data_types:
+            logging.info("IF Datastream found")
+            global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
+            if global_frame_body is None:
+                return None
+            return AmmosSingleFrame(global_frame_header, global_frame_body)
+
+        if global_frame_body is None:
+            return None
+
+        logging.info("Unsupported frame type %s found", global_frame_header.frame_type)
+        return None
+
+

Ancestors

+
    +
  • abc.ABC
    • +
+

Subclasses

+ +

Instance variables

+
+
var ammos_file
+
+

I return the file to read the data from.

+
+ +Expand source code + +
@property
+def ammos_file(self):
+    """I return the file to read the data from."""
+    return self.__ammos_file
+
+
+
var container
+
+

I return the container which stores the data read.

+
+ +Expand source code + +
@property
+def container(self):
+    """I return the container which stores the data read."""
+    return self.__container
+
+
+
var file_name
+
+

I return the name of the original file.

+
+ +Expand source code + +
@property
+def file_name(self):
+    """I return the name of the original file."""
+    return self.__file_name
+
+
+
var tags
+
+

I return all the tags.

+
+ +Expand source code + +
@property
+def tags(self):
+    """I return all the tags."""
+    return self.__tags
+
+
+
+

Methods

+
+
+def add_tag(self, a_key, a_value) +
+
+

I add information to tags using a key/value pair.

+
+ +Expand source code + +
def add_tag(self, a_key, a_value):
+    """I add information to tags using a key/value pair."""
+    assert a_key not in self.__tags
+    self.__tags[a_key] = a_value
+
+
+
+def read_all_frames_left(self) +
+
+

I read all remaining frames into my container until end of file is reached.

+

:return: a container containing all frames read +:rtype: AmmosContainer

+
+ +Expand source code + +
def read_all_frames_left(self):
+    """
+    I read all remaining frames into my container until end of file is reached.
+
+    :return: a container containing all frames read
+    :rtype: AmmosContainer
+    """
+    frames_read = 0
+    while True:
+        logging.info("Reading single frame %s ...", frames_read)
+        current_frame = self.read_next_single_frame()
+        if current_frame is not None:
+            frames_read += 1
+            self.container.add_frame(current_frame)
+        else:
+            logging.debug("Frame: %s incomplete", frames_read+1)
+            break
+    logging.info("%s frames read", len(self.container.global_frames))
+    return self.container
+
+
+
+def read_next_global_frame_body(self, data_header_length) +
+
+

My descendents have to implement this.

+
+ +Expand source code + +
@abstractmethod
+def read_next_global_frame_body(self, data_header_length):
+    """My descendents have to implement this."""
+    pass
+
+
+
+def read_next_global_frame_header(self) +
+
+

I return the next global frame header read from current position in file.

+

:return: the next global frame header or None if incomplete +:rtype: AmmosGlobalFrameHeader

+
+ +Expand source code + +
def read_next_global_frame_header(self):
+    """
+    I return the next global frame header read from current position in file.
+
+    :return: the next global frame header or None if incomplete
+    :rtype: AmmosGlobalFrameHeader
+    """
+    header_size = AmmosGlobalFrameHeader.HEADER_SIZE
+
+    bytes = self.ammos_file.read(header_size)
+    logging.info("Reading next global frame header")
+    if ((not bytes) or (len(bytes) < header_size)):
+        logging.debug("Can not read all %s bytes of global frame header", header_size)
+        return None
+
+    current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes)
+
+    if current_global_frame_header is None:
+        return None
+
+    # logging.info("Current global frame header %s", current_global_frame_header)
+    return current_global_frame_header
+
+
+
+def read_next_single_frame(self) +
+
+

I read and return a single global frame.

+

:return: a single global frame +:rtype: AmmosSingleFrame

+
+ +Expand source code + +
def read_next_single_frame(self):
+    """
+    I read and return a single global frame.
+
+    :return: a single global frame
+    :rtype: AmmosSingleFrame
+    """
+    global_frame_header = self.read_next_global_frame_header()
+
+    if global_frame_header is None:
+        return None
+
+    if global_frame_header is None:
+        logging.debug("Global frame header missing")
+        return None
+
+    if global_frame_header.data_header_length is None:
+        logging.debug("Data header length empty")
+        return None
+
+    if global_frame_header.frame_type not in list(FrameType):
+        logging.info("Unknown frame type %s found", global_frame_header.frame_type)
+        return None
+
+    # FIXME: Refactor duplicate logic - This code stinks
+
+    global_frame_body = None
+
+    if_data_types = [FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT.value,
+                     FrameType.IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT.value,
+                     FrameType.IF_DATA_16BIT_REAL_REAL.value,
+                     FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED.value,
+                     FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED.value]
+
+    if global_frame_header.frame_type == FrameType.AUDIO_DATA.value:
+        logging.info("Audio Datastream found")
+        global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
+        if global_frame_body is None:
+            return None
+        return AmmosSingleFrame(global_frame_header, global_frame_body)
+
+    if global_frame_header.frame_type in if_data_types:
+        logging.info("IF Datastream found")
+        global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length)
+        if global_frame_body is None:
+            return None
+        return AmmosSingleFrame(global_frame_header, global_frame_body)
+
+    if global_frame_body is None:
+        return None
+
+    logging.info("Unsupported frame type %s found", global_frame_header.frame_type)
+    return None
+
+
+
+def rewind_to_start(self) +
+
+

I set the file pointer to the beginning of the file for the next operation.

+
+ +Expand source code + +
def rewind_to_start(self):
+    """I set the file pointer to the beginning of the file for the next operation."""
+    self.ammos_file.seek(0)
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosAudioDataBody.html =================================================================== --- doc/ammosreader/AmmosAudioDataBody.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosAudioDataBody.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,190 @@ + + + + + + +ammosreader.AmmosAudioDataBody API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosAudioDataBody

+
+
+

I provide an AMMOS data block for audio data frames.

+
+ +Expand source code + +
"""I provide an AMMOS data block for audio data frames."""
+
+
+class AmmosAudioDataBody:
+    """I implement an AMMOS data block for audio data frames."""
+
+    def __init__(self, pcm_data, number_of_channels=1, samples_per_channel=1, sample_size=1):
+        """I return a new AMMOS data block for audio data frames."""
+        self.__number_of_channels = number_of_channels
+        self.__samples_per_channel = samples_per_channel
+        self.__sample_size = sample_size
+        self.__payload = pcm_data
+
+    @property
+    def payload(self):
+        """I return the raw pcm data with channels interweaved."""
+        return self.__payload
+
+    def pcm_for_channel(self, channel_number):
+        """I return the raw pcm data for a given channel."""
+        start_offset = channel_number * self.__sample_size
+        step = self.__sample_size * self.__number_of_channels
+        end = (len(self.__payload) // step) * step
+        channel_bytes = b""
+        for each in range(start_offset, end, step):
+            channel_bytes += self.__payload[each:each+self.__sample_size]
+        return channel_bytes
+
+    def __str__(self):
+        """I return the string representation of myself."""
+        return ("Number of channels:" + str(self.__number_of_channels) +
+                "\nSamples per channel:" + str(self.__samples_per_channel) +
+                "\nSample size:" + str(self.__sample_size) +
+                "\nData size:" + str(len(self.__payload)))
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosAudioDataBody +(pcm_data, number_of_channels=1, samples_per_channel=1, sample_size=1) +
+
+

I implement an AMMOS data block for audio data frames.

+

I return a new AMMOS data block for audio data frames.

+
+ +Expand source code + +
class AmmosAudioDataBody:
+    """I implement an AMMOS data block for audio data frames."""
+
+    def __init__(self, pcm_data, number_of_channels=1, samples_per_channel=1, sample_size=1):
+        """I return a new AMMOS data block for audio data frames."""
+        self.__number_of_channels = number_of_channels
+        self.__samples_per_channel = samples_per_channel
+        self.__sample_size = sample_size
+        self.__payload = pcm_data
+
+    @property
+    def payload(self):
+        """I return the raw pcm data with channels interweaved."""
+        return self.__payload
+
+    def pcm_for_channel(self, channel_number):
+        """I return the raw pcm data for a given channel."""
+        start_offset = channel_number * self.__sample_size
+        step = self.__sample_size * self.__number_of_channels
+        end = (len(self.__payload) // step) * step
+        channel_bytes = b""
+        for each in range(start_offset, end, step):
+            channel_bytes += self.__payload[each:each+self.__sample_size]
+        return channel_bytes
+
+    def __str__(self):
+        """I return the string representation of myself."""
+        return ("Number of channels:" + str(self.__number_of_channels) +
+                "\nSamples per channel:" + str(self.__samples_per_channel) +
+                "\nSample size:" + str(self.__sample_size) +
+                "\nData size:" + str(len(self.__payload)))
+
+

Instance variables

+
+
var payload
+
+

I return the raw pcm data with channels interweaved.

+
+ +Expand source code + +
@property
+def payload(self):
+    """I return the raw pcm data with channels interweaved."""
+    return self.__payload
+
+
+
+

Methods

+
+
+def pcm_for_channel(self, channel_number) +
+
+

I return the raw pcm data for a given channel.

+
+ +Expand source code + +
def pcm_for_channel(self, channel_number):
+    """I return the raw pcm data for a given channel."""
+    start_offset = channel_number * self.__sample_size
+    step = self.__sample_size * self.__number_of_channels
+    end = (len(self.__payload) // step) * step
+    channel_bytes = b""
+    for each in range(start_offset, end, step):
+        channel_bytes += self.__payload[each:each+self.__sample_size]
+    return channel_bytes
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosAudioDataHeader.html =================================================================== --- doc/ammosreader/AmmosAudioDataHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosAudioDataHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,216 @@ + + + + + + +ammosreader.AmmosAudioDataHeader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosAudioDataHeader

+
+
+

I provide an AMMOS data header for audio data frames.

+
+ +Expand source code + +
"""I provide an AMMOS data header for audio data frames."""
+
+import struct
+from ammosreader.AmmosConstants import AmmosAudioDemodType
+
+
+class AmmosAudioDataHeader:
+    """I implement an AMMOS data header for audio data frames."""
+
+    HEADER_SIZE = 36  # 9 words
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return an AMMOS data header from given bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+        elements = struct.unpack('<IIQIIIII', bytes)
+        sample_rate = elements[0]
+        status = elements[1]
+        frequency = elements[2]
+        demod_bandwidth = elements[3]
+        demod_type = AmmosAudioDemodType(elements[4])
+        number_of_samples = elements[5]
+        number_of_channels = elements[6]
+        sample_size = elements[7]
+        return AmmosAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
+                                    number_of_samples, number_of_channels, sample_size)
+
+    def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples,
+                 number_of_channels, sample_size):
+        """I create a new instance of myself using the above parameters."""
+        self.sample_rate = sample_rate
+        self.status = status
+        self.frequency = frequency
+        self.demod_bandwidth = demod_bandwidth
+        self.demod_type = AmmosAudioDemodType(demod_type)
+        self.number_of_samples = number_of_samples
+        self.number_of_channels = number_of_channels
+        self.sample_size = sample_size
+
+    def __str__(self):
+        """I return the string representation of myself."""
+        return ("\nAmmosAudioDataHeader\n" +
+                "Sample rate:" + str(self.sample_rate) + "\n" +
+                "Status:" + str(self.status) + "\n" +
+                "Frequency:" + str(self.frequency) + "\n" +
+                "Demodulation bandwidth:" + str(self.demod_bandwidth) + "\n" +
+                "Demodulation type:" + str(self.demod_type) + "\n" +
+                "Sample count:" + str(self.number_of_samples) + "\n" +
+                "Channel count:" + str(self.number_of_channels) + "\n" +
+                "Sample size:" + str(self.sample_size) + "\n")
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosAudioDataHeader +(sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples, number_of_channels, sample_size) +
+
+

I implement an AMMOS data header for audio data frames.

+

I create a new instance of myself using the above parameters.

+
+ +Expand source code + +
class AmmosAudioDataHeader:
+    """I implement an AMMOS data header for audio data frames."""
+
+    HEADER_SIZE = 36  # 9 words
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return an AMMOS data header from given bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+        elements = struct.unpack('<IIQIIIII', bytes)
+        sample_rate = elements[0]
+        status = elements[1]
+        frequency = elements[2]
+        demod_bandwidth = elements[3]
+        demod_type = AmmosAudioDemodType(elements[4])
+        number_of_samples = elements[5]
+        number_of_channels = elements[6]
+        sample_size = elements[7]
+        return AmmosAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
+                                    number_of_samples, number_of_channels, sample_size)
+
+    def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples,
+                 number_of_channels, sample_size):
+        """I create a new instance of myself using the above parameters."""
+        self.sample_rate = sample_rate
+        self.status = status
+        self.frequency = frequency
+        self.demod_bandwidth = demod_bandwidth
+        self.demod_type = AmmosAudioDemodType(demod_type)
+        self.number_of_samples = number_of_samples
+        self.number_of_channels = number_of_channels
+        self.sample_size = sample_size
+
+    def __str__(self):
+        """I return the string representation of myself."""
+        return ("\nAmmosAudioDataHeader\n" +
+                "Sample rate:" + str(self.sample_rate) + "\n" +
+                "Status:" + str(self.status) + "\n" +
+                "Frequency:" + str(self.frequency) + "\n" +
+                "Demodulation bandwidth:" + str(self.demod_bandwidth) + "\n" +
+                "Demodulation type:" + str(self.demod_type) + "\n" +
+                "Sample count:" + str(self.number_of_samples) + "\n" +
+                "Channel count:" + str(self.number_of_channels) + "\n" +
+                "Sample size:" + str(self.sample_size) + "\n")
+
+

Class variables

+
+
var HEADER_SIZE
+
+
+
+
+

Static methods

+
+
+def from_bytes(bytes) +
+
+

I return an AMMOS data header from given bytes.

+
+ +Expand source code + +
@classmethod
+def from_bytes(cls, bytes):
+    """I return an AMMOS data header from given bytes."""
+    assert len(bytes) == cls.HEADER_SIZE
+    elements = struct.unpack('<IIQIIIII', bytes)
+    sample_rate = elements[0]
+    status = elements[1]
+    frequency = elements[2]
+    demod_bandwidth = elements[3]
+    demod_type = AmmosAudioDemodType(elements[4])
+    number_of_samples = elements[5]
+    number_of_channels = elements[6]
+    sample_size = elements[7]
+    return AmmosAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
+                                number_of_samples, number_of_channels, sample_size)
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosAudioReader.html =================================================================== --- doc/ammosreader/AmmosAudioReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosAudioReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,525 @@ + + + + + + +ammosreader.AmmosAudioReader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosAudioReader

+
+
+

I provide a specialized Ammos Reader for audio data.

+
+ +Expand source code + +
"""I provide a specialized Ammos Reader for audio data."""
+import logging
+
+from ammosreader.AbstractAmmosReader import AbstractAmmosReader
+from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody
+from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
+from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader
+from ammosreader.AmmosAudioDataBody import AmmosAudioDataBody
+
+
+class AmmosAudioReader(AbstractAmmosReader):
+    """I read the audio data embedded in an R&S AMMOS recording."""
+
+    def __init__(self, file_name):
+        """
+        I return an instance of AmmosAudioReader initialized with a given file name.
+
+        :param file_name: the file to read from
+        :type file_name: str
+        """
+        super().__init__(file_name)
+
+    def read_next_global_frame_body_data_header(self):
+        """
+        I return the next global frame body data header from current position in file.
+
+        :param data_header_size: the number of bytes to read
+        :type data_header_size: int
+        :return: the next Ammos Audio Data header or None if incomplete
+        :rtype: AmmosAudioDataHeader
+        """
+        header_size = AmmosAudioDataHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+
+        logging.info("\nReading global frame body standard data header\n")
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame body data header", header_size)
+            return None
+        return AmmosAudioDataHeader.from_bytes(bytes)
+
+    def read_next_global_frame_body_extended_data_header(self):
+        """
+        I return the next global frame body extended data header from current position in file.
+
+        :return: the next Ammos Audio Extended Data header or None if incomplete
+        :rtype: AmmosExtendedAudioDataHeader
+        """
+        header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
+            return None
+        return AmmosExtendedAudioDataHeader.from_bytes(bytes)
+
+    def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
+        """
+        I return the next audio data read from current position in file.
+
+        :param number_of_samples: the number of samples per channel inside data body
+        :type number_of_samples: int
+
+        :param number_of_channels: number of channels (e.g. mono, stereo or even more)
+        :type number_of_channels: int
+
+        :param sample_size: sample size in bytes (1, 2 or 4 bytes)
+        :type sample_size: int
+
+        :return: the next audio data or None if incomplete
+        :rtype: bytes
+        """
+        total = number_of_samples*number_of_channels*sample_size
+
+        byte_string = self.ammos_file.read(total)
+
+        if len(byte_string) != total:
+            logging.debug("Can not read all %s bytes of data body", total)
+            return None
+        return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)
+
+    def read_next_global_frame_body(self, data_header_length):
+        """
+        I return the next global frame body read from current position in file.
+
+        :param data_header_length: the length of the data header
+        :type data_header_length: int
+        """
+        audio_data_header = None
+
+        if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
+            logging.info("Read standard data header")
+            audio_data_header = self.read_next_global_frame_body_data_header()
+
+        logging.info("Data header length %s", data_header_length)
+        if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
+            logging.info("Read extended data header")
+            audio_data_header = self.read_next_global_frame_body_extended_data_header()
+
+        if audio_data_header is None:
+            logging.debug("Data header missing or format unknown")
+            return None
+
+        audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
+                                                         audio_data_header.number_of_channels,
+                                                         audio_data_header.sample_size)
+
+        if audio_data_body is None:
+            logging.debug("Data body missing")
+            return None
+
+        return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
+
+    def pcm_for_channel(self, a_channel):
+        """
+        I return the raw pcm audio data for a given channel.
+
+        :param a_channel: the channel I have to extract
+        :type a_channel: int
+
+        :rtype: bytes
+        """
+        return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosAudioReader +(file_name) +
+
+

I read the audio data embedded in an R&S AMMOS recording.

+

I return an instance of AmmosAudioReader initialized with a given file name.

+

:param file_name: the file to read from +:type file_name: str

+
+ +Expand source code + +
class AmmosAudioReader(AbstractAmmosReader):
+    """I read the audio data embedded in an R&S AMMOS recording."""
+
+    def __init__(self, file_name):
+        """
+        I return an instance of AmmosAudioReader initialized with a given file name.
+
+        :param file_name: the file to read from
+        :type file_name: str
+        """
+        super().__init__(file_name)
+
+    def read_next_global_frame_body_data_header(self):
+        """
+        I return the next global frame body data header from current position in file.
+
+        :param data_header_size: the number of bytes to read
+        :type data_header_size: int
+        :return: the next Ammos Audio Data header or None if incomplete
+        :rtype: AmmosAudioDataHeader
+        """
+        header_size = AmmosAudioDataHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+
+        logging.info("\nReading global frame body standard data header\n")
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame body data header", header_size)
+            return None
+        return AmmosAudioDataHeader.from_bytes(bytes)
+
+    def read_next_global_frame_body_extended_data_header(self):
+        """
+        I return the next global frame body extended data header from current position in file.
+
+        :return: the next Ammos Audio Extended Data header or None if incomplete
+        :rtype: AmmosExtendedAudioDataHeader
+        """
+        header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
+            return None
+        return AmmosExtendedAudioDataHeader.from_bytes(bytes)
+
+    def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
+        """
+        I return the next audio data read from current position in file.
+
+        :param number_of_samples: the number of samples per channel inside data body
+        :type number_of_samples: int
+
+        :param number_of_channels: number of channels (e.g. mono, stereo or even more)
+        :type number_of_channels: int
+
+        :param sample_size: sample size in bytes (1, 2 or 4 bytes)
+        :type sample_size: int
+
+        :return: the next audio data or None if incomplete
+        :rtype: bytes
+        """
+        total = number_of_samples*number_of_channels*sample_size
+
+        byte_string = self.ammos_file.read(total)
+
+        if len(byte_string) != total:
+            logging.debug("Can not read all %s bytes of data body", total)
+            return None
+        return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)
+
+    def read_next_global_frame_body(self, data_header_length):
+        """
+        I return the next global frame body read from current position in file.
+
+        :param data_header_length: the length of the data header
+        :type data_header_length: int
+        """
+        audio_data_header = None
+
+        if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
+            logging.info("Read standard data header")
+            audio_data_header = self.read_next_global_frame_body_data_header()
+
+        logging.info("Data header length %s", data_header_length)
+        if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
+            logging.info("Read extended data header")
+            audio_data_header = self.read_next_global_frame_body_extended_data_header()
+
+        if audio_data_header is None:
+            logging.debug("Data header missing or format unknown")
+            return None
+
+        audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
+                                                         audio_data_header.number_of_channels,
+                                                         audio_data_header.sample_size)
+
+        if audio_data_body is None:
+            logging.debug("Data body missing")
+            return None
+
+        return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
+
+    def pcm_for_channel(self, a_channel):
+        """
+        I return the raw pcm audio data for a given channel.
+
+        :param a_channel: the channel I have to extract
+        :type a_channel: int
+
+        :rtype: bytes
+        """
+        return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])
+
+

Ancestors

+ +

Methods

+
+
+def pcm_for_channel(self, a_channel) +
+
+

I return the raw pcm audio data for a given channel.

+

:param a_channel: the channel I have to extract +:type a_channel: int

+

:rtype: bytes

+
+ +Expand source code + +
def pcm_for_channel(self, a_channel):
+    """
+    I return the raw pcm audio data for a given channel.
+
+    :param a_channel: the channel I have to extract
+    :type a_channel: int
+
+    :rtype: bytes
+    """
+    return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])
+
+
+
+def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size) +
+
+

I return the next audio data read from current position in file.

+

:param number_of_samples: the number of samples per channel inside data body +:type number_of_samples: int

+

:param number_of_channels: number of channels (e.g. mono, stereo or even more) +:type number_of_channels: int

+

:param sample_size: sample size in bytes (1, 2 or 4 bytes) +:type sample_size: int

+

:return: the next audio data or None if incomplete +:rtype: bytes

+
+ +Expand source code + +
def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
+    """
+    I return the next audio data read from current position in file.
+
+    :param number_of_samples: the number of samples per channel inside data body
+    :type number_of_samples: int
+
+    :param number_of_channels: number of channels (e.g. mono, stereo or even more)
+    :type number_of_channels: int
+
+    :param sample_size: sample size in bytes (1, 2 or 4 bytes)
+    :type sample_size: int
+
+    :return: the next audio data or None if incomplete
+    :rtype: bytes
+    """
+    total = number_of_samples*number_of_channels*sample_size
+
+    byte_string = self.ammos_file.read(total)
+
+    if len(byte_string) != total:
+        logging.debug("Can not read all %s bytes of data body", total)
+        return None
+    return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)
+
+
+
+def read_next_global_frame_body(self, data_header_length) +
+
+

I return the next global frame body read from current position in file.

+

:param data_header_length: the length of the data header +:type data_header_length: int

+
+ +Expand source code + +
def read_next_global_frame_body(self, data_header_length):
+    """
+    I return the next global frame body read from current position in file.
+
+    :param data_header_length: the length of the data header
+    :type data_header_length: int
+    """
+    audio_data_header = None
+
+    if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
+        logging.info("Read standard data header")
+        audio_data_header = self.read_next_global_frame_body_data_header()
+
+    logging.info("Data header length %s", data_header_length)
+    if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
+        logging.info("Read extended data header")
+        audio_data_header = self.read_next_global_frame_body_extended_data_header()
+
+    if audio_data_header is None:
+        logging.debug("Data header missing or format unknown")
+        return None
+
+    audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
+                                                     audio_data_header.number_of_channels,
+                                                     audio_data_header.sample_size)
+
+    if audio_data_body is None:
+        logging.debug("Data body missing")
+        return None
+
+    return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
+
+
+
+def read_next_global_frame_body_data_header(self) +
+
+

I return the next global frame body data header from current position in file.

+

:param data_header_size: the number of bytes to read +:type data_header_size: int +:return: the next Ammos Audio Data header or None if incomplete +:rtype: AmmosAudioDataHeader

+
+ +Expand source code + +
def read_next_global_frame_body_data_header(self):
+    """
+    I return the next global frame body data header from current position in file.
+
+    :param data_header_size: the number of bytes to read
+    :type data_header_size: int
+    :return: the next Ammos Audio Data header or None if incomplete
+    :rtype: AmmosAudioDataHeader
+    """
+    header_size = AmmosAudioDataHeader.HEADER_SIZE
+
+    bytes = self.ammos_file.read(header_size)
+
+    logging.info("\nReading global frame body standard data header\n")
+    if ((not bytes) or (len(bytes) < header_size)):
+        logging.debug("Can not read all %s bytes of global frame body data header", header_size)
+        return None
+    return AmmosAudioDataHeader.from_bytes(bytes)
+
+
+
+def read_next_global_frame_body_extended_data_header(self) +
+
+

I return the next global frame body extended data header from current position in file.

+

:return: the next Ammos Audio Extended Data header or None if incomplete +:rtype: AmmosExtendedAudioDataHeader

+
+ +Expand source code + +
def read_next_global_frame_body_extended_data_header(self):
+    """
+    I return the next global frame body extended data header from current position in file.
+
+    :return: the next Ammos Audio Extended Data header or None if incomplete
+    :rtype: AmmosExtendedAudioDataHeader
+    """
+    header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
+
+    bytes = self.ammos_file.read(header_size)
+
+    if ((not bytes) or (len(bytes) < header_size)):
+        logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
+        return None
+    return AmmosExtendedAudioDataHeader.from_bytes(bytes)
+
+
+
+

Inherited members

+ +
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosAudioSocketReader.html =================================================================== --- doc/ammosreader/AmmosAudioSocketReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosAudioSocketReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,497 @@ + + + + + + +ammosreader.AmmosAudioSocketReader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosAudioSocketReader

+
+
+

I read a Ammos datastream from a socket.

+
+ +Expand source code + +
"""I read a Ammos datastream from a socket."""
+
+import select
+import socket
+from collections import deque
+import numpy as np
+from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
+from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader
+from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader
+
+
+class AmmosAudioSocketReader:
+    def __init__(self, socket:socket.socket, debug=True):
+        """
+        Initializes the AmmosAudioSocketReader
+
+        Args:
+            socket (socket.socket): socket to read from
+            debug (bool): if true, prints debug information
+        """
+
+        #buffer for reading socket bytewise und check for the magic word
+        self.__magic_word_buffer = deque(maxlen=4)
+
+        #input socket to read from
+        self.__socket = socket
+
+        #
+        self.DEBUG_MODE = debug
+
+    def __get_next_data(self, byte_count: int) -> bytearray:
+        """
+        Gets the next bytes from the socket, for example headers and body data.
+
+        Args:
+            byte_count (int): number of bytes to read
+
+        Raises:
+            TimeoutError: Raises TimeoutError if the socket does not serve data anymore
+
+        Returns:
+            bytearray: data from socket as bytearray
+        """
+
+        byte_array = []
+        
+        while len(b''.join(byte_array)) < byte_count:
+            if self.DEBUG_MODE:
+                print(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}")
+            self.__socket.settimeout(5)
+            new_bytes = self.__socket.recv(byte_count - len(b''.join(byte_array)), socket.MSG_WAITALL)
+
+            if not new_bytes:
+                raise TimeoutError("Socket timed out while reading data")
+
+            if self.DEBUG_MODE:
+                print(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} ramining")
+
+            byte_array.append(new_bytes)
+
+        return b''.join(byte_array)
+
+    def __audio_data_body_to_numpy(self, audio_data_body:bytearray) -> np.ndarray:
+        """
+        converts the audio data body to a numpy array
+
+        Args:
+            audio_data_body (bytearray): audio data from audio data body
+
+        Returns:
+            np.ndarray: audio data as numpy array
+        """
+
+        return np.frombuffer(audio_data_body, dtype=np.int16)
+
+    def read_next_frame(self) -> tuple[np.ndarray, int]:
+        """Reads the next ammos audio frame from socket
+
+        Raises:
+            TimeoutError: Raisees TimeoutError if the socket does not serve data anymore
+
+        Returns:
+            tuple[np.ndarray, int]: Contains the audio data and the sample rate
+        """
+
+        # get first byte of the day
+        self.__socket.settimeout(5)
+
+        new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
+        # raise Exception if socket does not return anything
+        if len(new_byte) < 1:
+            raise TimeoutError      
+
+        #read loop
+        while new_byte:
+            #
+            self.__magic_word_buffer.append(new_byte)
+            byte_array = b''.join(self.__magic_word_buffer)
+
+            if byte_array.hex() == '726574fb':
+                #print(byte_array.hex())
+
+                ammos_global_header_buffer = list(self.__magic_word_buffer)
+                ammos_global_header_buffer.append(self.__get_next_data(20))
+                #while len(b''.join(ammos_global_header_buffer)) < 24:
+                #    ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer))))
+                    
+                ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer))
+                if self.DEBUG_MODE:
+                    print(ammos_global_header)
+
+                if ammos_global_header.data_header_length == 44 and ammos_global_header.frame_type == 256:
+                    byte_array_header = self.__get_next_data(44)
+                    #while len(b''.join(byte_array_header)) < 44:
+                    #    byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header))))
+
+                    ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header)
+                    if self.DEBUG_MODE:
+                        print(ammos_extended_audio_data_header.number_of_samples, ammos_extended_audio_data_header.number_of_channels, ammos_extended_audio_data_header.sample_size)
+                    audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples* 
+                                                      ammos_extended_audio_data_header.number_of_channels* 
+                                                      ammos_extended_audio_data_header.sample_size)
+
+                    audio_array = self.__audio_data_body_to_numpy(audio_body)
+                    if self.DEBUG_MODE:
+                        print(len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
+
+                    return [audio_array, ammos_extended_audio_data_header.sample_rate]
+
+                elif ammos_global_header.data_header_length == 36 and ammos_global_header.frame_type == 256:
+                    byte_array_header = self.__get_next_data(36)
+                    #while len(b''.join(byte_array_header)) < 36:
+                    #    byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header))))
+
+                    ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header)
+                    if self.DEBUG_MODE:
+                        print(ammos_audio_data_header.number_of_samples, ammos_audio_data_header.number_of_channels, ammos_audio_data_header.sample_size)
+                    audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples* 
+                                                      ammos_extended_audio_data_header.number_of_channels* 
+                                                      ammos_extended_audio_data_header.sample_size)
+
+                    audio_array = self.__audio_data_body_to_numpy(audio_body)
+                    if self.DEBUG_MODE:
+                        print(len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
+
+                    return [audio_array, ammos_audio_data_header.sample_rate]
+
+            # get the next byte
+            self.__socket.settimeout(5)
+
+            new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
+            # raise Exception if socket does not return anything
+            if len(new_byte) < 1:
+                raise TimeoutError   
+
+        return None
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosAudioSocketReader +(socket: socket.socket, debug=True) +
+
+

Initializes the AmmosAudioSocketReader

+

Args

+
+
socket : socket.socket
+
socket to read from
+
debug : bool
+
if true, prints debug information
+
+
+ +Expand source code + +
class AmmosAudioSocketReader:
+    def __init__(self, socket:socket.socket, debug=True):
+        """
+        Initializes the AmmosAudioSocketReader
+
+        Args:
+            socket (socket.socket): socket to read from
+            debug (bool): if true, prints debug information
+        """
+
+        #buffer for reading socket bytewise und check for the magic word
+        self.__magic_word_buffer = deque(maxlen=4)
+
+        #input socket to read from
+        self.__socket = socket
+
+        #
+        self.DEBUG_MODE = debug
+
+    def __get_next_data(self, byte_count: int) -> bytearray:
+        """
+        Gets the next bytes from the socket, for example headers and body data.
+
+        Args:
+            byte_count (int): number of bytes to read
+
+        Raises:
+            TimeoutError: Raises TimeoutError if the socket does not serve data anymore
+
+        Returns:
+            bytearray: data from socket as bytearray
+        """
+
+        byte_array = []
+        
+        while len(b''.join(byte_array)) < byte_count:
+            if self.DEBUG_MODE:
+                print(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}")
+            self.__socket.settimeout(5)
+            new_bytes = self.__socket.recv(byte_count - len(b''.join(byte_array)), socket.MSG_WAITALL)
+
+            if not new_bytes:
+                raise TimeoutError("Socket timed out while reading data")
+
+            if self.DEBUG_MODE:
+                print(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} ramining")
+
+            byte_array.append(new_bytes)
+
+        return b''.join(byte_array)
+
+    def __audio_data_body_to_numpy(self, audio_data_body:bytearray) -> np.ndarray:
+        """
+        converts the audio data body to a numpy array
+
+        Args:
+            audio_data_body (bytearray): audio data from audio data body
+
+        Returns:
+            np.ndarray: audio data as numpy array
+        """
+
+        return np.frombuffer(audio_data_body, dtype=np.int16)
+
+    def read_next_frame(self) -> tuple[np.ndarray, int]:
+        """Reads the next ammos audio frame from socket
+
+        Raises:
+            TimeoutError: Raisees TimeoutError if the socket does not serve data anymore
+
+        Returns:
+            tuple[np.ndarray, int]: Contains the audio data and the sample rate
+        """
+
+        # get first byte of the day
+        self.__socket.settimeout(5)
+
+        new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
+        # raise Exception if socket does not return anything
+        if len(new_byte) < 1:
+            raise TimeoutError      
+
+        #read loop
+        while new_byte:
+            #
+            self.__magic_word_buffer.append(new_byte)
+            byte_array = b''.join(self.__magic_word_buffer)
+
+            if byte_array.hex() == '726574fb':
+                #print(byte_array.hex())
+
+                ammos_global_header_buffer = list(self.__magic_word_buffer)
+                ammos_global_header_buffer.append(self.__get_next_data(20))
+                #while len(b''.join(ammos_global_header_buffer)) < 24:
+                #    ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer))))
+                    
+                ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer))
+                if self.DEBUG_MODE:
+                    print(ammos_global_header)
+
+                if ammos_global_header.data_header_length == 44 and ammos_global_header.frame_type == 256:
+                    byte_array_header = self.__get_next_data(44)
+                    #while len(b''.join(byte_array_header)) < 44:
+                    #    byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header))))
+
+                    ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header)
+                    if self.DEBUG_MODE:
+                        print(ammos_extended_audio_data_header.number_of_samples, ammos_extended_audio_data_header.number_of_channels, ammos_extended_audio_data_header.sample_size)
+                    audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples* 
+                                                      ammos_extended_audio_data_header.number_of_channels* 
+                                                      ammos_extended_audio_data_header.sample_size)
+
+                    audio_array = self.__audio_data_body_to_numpy(audio_body)
+                    if self.DEBUG_MODE:
+                        print(len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
+
+                    return [audio_array, ammos_extended_audio_data_header.sample_rate]
+
+                elif ammos_global_header.data_header_length == 36 and ammos_global_header.frame_type == 256:
+                    byte_array_header = self.__get_next_data(36)
+                    #while len(b''.join(byte_array_header)) < 36:
+                    #    byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header))))
+
+                    ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header)
+                    if self.DEBUG_MODE:
+                        print(ammos_audio_data_header.number_of_samples, ammos_audio_data_header.number_of_channels, ammos_audio_data_header.sample_size)
+                    audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples* 
+                                                      ammos_extended_audio_data_header.number_of_channels* 
+                                                      ammos_extended_audio_data_header.sample_size)
+
+                    audio_array = self.__audio_data_body_to_numpy(audio_body)
+                    if self.DEBUG_MODE:
+                        print(len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
+
+                    return [audio_array, ammos_audio_data_header.sample_rate]
+
+            # get the next byte
+            self.__socket.settimeout(5)
+
+            new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
+            # raise Exception if socket does not return anything
+            if len(new_byte) < 1:
+                raise TimeoutError   
+
+        return None
+
+

Methods

+
+
+def read_next_frame(self) ‑> tuple[numpy.ndarray, int] +
+
+

Reads the next ammos audio frame from socket

+

Raises

+
+
TimeoutError
+
Raisees TimeoutError if the socket does not serve data anymore
+
+

Returns

+
+
tuple[np.ndarray, int]
+
Contains the audio data and the sample rate
+
+
+ +Expand source code + +
def read_next_frame(self) -> tuple[np.ndarray, int]:
+    """Reads the next ammos audio frame from socket
+
+    Raises:
+        TimeoutError: Raisees TimeoutError if the socket does not serve data anymore
+
+    Returns:
+        tuple[np.ndarray, int]: Contains the audio data and the sample rate
+    """
+
+    # get first byte of the day
+    self.__socket.settimeout(5)
+
+    new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
+    # raise Exception if socket does not return anything
+    if len(new_byte) < 1:
+        raise TimeoutError      
+
+    #read loop
+    while new_byte:
+        #
+        self.__magic_word_buffer.append(new_byte)
+        byte_array = b''.join(self.__magic_word_buffer)
+
+        if byte_array.hex() == '726574fb':
+            #print(byte_array.hex())
+
+            ammos_global_header_buffer = list(self.__magic_word_buffer)
+            ammos_global_header_buffer.append(self.__get_next_data(20))
+            #while len(b''.join(ammos_global_header_buffer)) < 24:
+            #    ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer))))
+                
+            ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer))
+            if self.DEBUG_MODE:
+                print(ammos_global_header)
+
+            if ammos_global_header.data_header_length == 44 and ammos_global_header.frame_type == 256:
+                byte_array_header = self.__get_next_data(44)
+                #while len(b''.join(byte_array_header)) < 44:
+                #    byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header))))
+
+                ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header)
+                if self.DEBUG_MODE:
+                    print(ammos_extended_audio_data_header.number_of_samples, ammos_extended_audio_data_header.number_of_channels, ammos_extended_audio_data_header.sample_size)
+                audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples* 
+                                                  ammos_extended_audio_data_header.number_of_channels* 
+                                                  ammos_extended_audio_data_header.sample_size)
+
+                audio_array = self.__audio_data_body_to_numpy(audio_body)
+                if self.DEBUG_MODE:
+                    print(len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
+
+                return [audio_array, ammos_extended_audio_data_header.sample_rate]
+
+            elif ammos_global_header.data_header_length == 36 and ammos_global_header.frame_type == 256:
+                byte_array_header = self.__get_next_data(36)
+                #while len(b''.join(byte_array_header)) < 36:
+                #    byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header))))
+
+                ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header)
+                if self.DEBUG_MODE:
+                    print(ammos_audio_data_header.number_of_samples, ammos_audio_data_header.number_of_channels, ammos_audio_data_header.sample_size)
+                audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples* 
+                                                  ammos_extended_audio_data_header.number_of_channels* 
+                                                  ammos_extended_audio_data_header.sample_size)
+
+                audio_array = self.__audio_data_body_to_numpy(audio_body)
+                if self.DEBUG_MODE:
+                    print(len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
+
+                return [audio_array, ammos_audio_data_header.sample_rate]
+
+        # get the next byte
+        self.__socket.settimeout(5)
+
+        new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
+        # raise Exception if socket does not return anything
+        if len(new_byte) < 1:
+            raise TimeoutError   
+
+    return None
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosConstants.html =================================================================== --- doc/ammosreader/AmmosConstants.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosConstants.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,436 @@ + + + + + + +ammosreader.AmmosConstants API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosConstants

+
+
+

I provide several constants used in R&S software.

+
+ +Expand source code + +
"""I provide several constants used in R&S software."""
+
+from enum import Enum
+
+
+class FrameType(int, Enum):
+    """I map numbers to human readable format specifiers."""
+
+    TEST_DATA = 0x00
+    IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT = 0x01
+    IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT = 0x02
+    IF_DATA_16BIT_REAL_REAL = 0x03
+    IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED = 0x04
+    IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED = 0x05
+    SPECTRUM_DATA_8BIT = 0x10
+    SPECTRUM_DATA_16BIT = 0x11
+    SPECTRUM_DATA_32BIT = 0x12
+    SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x13
+    SEGMENTATION_SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x14
+    HF_TUNING_INDICATOR_DATA = 0x20
+    HF_SCAN_CHANNEL_FOUND_DATA = 0x22
+    HF_SCAN_FREQUENCY_FOUND_DATA = 0x23
+    HF_SCAN_SWEEP_RESTARTED_DATA = 0x24
+    HF_DEMODULATION_SYMBOL_STREAM_DATA = 0x30
+    HF_DECODER_TEXT_DATA = 0x40
+    HF_SPECTRUM_VISUALIZATION_DATA = 0x50
+    HF_TIMEDOMAIN_VISUALIZATION_DATA = 0x51
+    AUDIO_DATA = 0x100
+    EMISSION_LIST_DATA = 0x110
+    LEVEL_DATA = 0x120
+    SYMBOL_DATA = 0x130
+    INSTANTANEOUS_DATA = 0x140
+    BURST_EMMISION_LIST = 0x150
+    IMAGE_DATA = 0x160
+    TRANSMISSION_SYSTEM_RESULT_DATA = 0x170
+    PULSE_DESCRIPTION_WORD_DATA = 0x200
+    PULSE_REPETITION_WORD_DATA = 0x210
+    EM050_SCAN_DATA = 0x4000
+    SCAN_LEVEL = 0x4001
+    SCAN_TUNING = 0x4002
+    SCAN_LEVEL_TUNING = 0x4003
+    DDF_RESERVED_START = 0x5000
+    DDF_RESERVED_END = 0x50FF
+
+
+class AmmosAudioDemodType(int, Enum):
+    """I map numbers to human readable demodulation types."""
+
+    FM = 0
+    AM = 1
+    ISB = 5
+    CW = 6
+    USB = 7
+    LSB = 8
+    DIGITAL = 256
+    UNKNOWN = 0xFFFFFFFF
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosAudioDemodType +(value, names=None, *, module=None, qualname=None, type=None, start=1) +
+
+

I map numbers to human readable demodulation types.

+
+ +Expand source code + +
class AmmosAudioDemodType(int, Enum):
+    """I map numbers to human readable demodulation types."""
+
+    FM = 0
+    AM = 1
+    ISB = 5
+    CW = 6
+    USB = 7
+    LSB = 8
+    DIGITAL = 256
+    UNKNOWN = 0xFFFFFFFF
+
+

Ancestors

+
    +
  • builtins.int
    • +
  • enum.Enum
    • +
+

Class variables

+
+
var AM
+
+
+
+
var CW
+
+
+
+
var DIGITAL
+
+
+
+
var FM
+
+
+
+
var ISB
+
+
+
+
var LSB
+
+
+
+
var UNKNOWN
+
+
+
+
var USB
+
+
+
+
+
+
+class FrameType +(value, names=None, *, module=None, qualname=None, type=None, start=1) +
+
+

I map numbers to human readable format specifiers.

+
+ +Expand source code + +
class FrameType(int, Enum):
+    """I map numbers to human readable format specifiers."""
+
+    TEST_DATA = 0x00
+    IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT = 0x01
+    IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT = 0x02
+    IF_DATA_16BIT_REAL_REAL = 0x03
+    IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED = 0x04
+    IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED = 0x05
+    SPECTRUM_DATA_8BIT = 0x10
+    SPECTRUM_DATA_16BIT = 0x11
+    SPECTRUM_DATA_32BIT = 0x12
+    SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x13
+    SEGMENTATION_SPECTRUM_DATA_32BIT_FLOATINGPOINT = 0x14
+    HF_TUNING_INDICATOR_DATA = 0x20
+    HF_SCAN_CHANNEL_FOUND_DATA = 0x22
+    HF_SCAN_FREQUENCY_FOUND_DATA = 0x23
+    HF_SCAN_SWEEP_RESTARTED_DATA = 0x24
+    HF_DEMODULATION_SYMBOL_STREAM_DATA = 0x30
+    HF_DECODER_TEXT_DATA = 0x40
+    HF_SPECTRUM_VISUALIZATION_DATA = 0x50
+    HF_TIMEDOMAIN_VISUALIZATION_DATA = 0x51
+    AUDIO_DATA = 0x100
+    EMISSION_LIST_DATA = 0x110
+    LEVEL_DATA = 0x120
+    SYMBOL_DATA = 0x130
+    INSTANTANEOUS_DATA = 0x140
+    BURST_EMMISION_LIST = 0x150
+    IMAGE_DATA = 0x160
+    TRANSMISSION_SYSTEM_RESULT_DATA = 0x170
+    PULSE_DESCRIPTION_WORD_DATA = 0x200
+    PULSE_REPETITION_WORD_DATA = 0x210
+    EM050_SCAN_DATA = 0x4000
+    SCAN_LEVEL = 0x4001
+    SCAN_TUNING = 0x4002
+    SCAN_LEVEL_TUNING = 0x4003
+    DDF_RESERVED_START = 0x5000
+    DDF_RESERVED_END = 0x50FF
+
+

Ancestors

+
    +
  • builtins.int
    • +
  • enum.Enum
    • +
+

Class variables

+
+
var AUDIO_DATA
+
+
+
+
var BURST_EMMISION_LIST
+
+
+
+
var DDF_RESERVED_END
+
+
+
+
var DDF_RESERVED_START
+
+
+
+
var EM050_SCAN_DATA
+
+
+
+
var EMISSION_LIST_DATA
+
+
+
+
var HF_DECODER_TEXT_DATA
+
+
+
+
var HF_DEMODULATION_SYMBOL_STREAM_DATA
+
+
+
+
var HF_SCAN_CHANNEL_FOUND_DATA
+
+
+
+
var HF_SCAN_FREQUENCY_FOUND_DATA
+
+
+
+
var HF_SCAN_SWEEP_RESTARTED_DATA
+
+
+
+
var HF_SPECTRUM_VISUALIZATION_DATA
+
+
+
+
var HF_TIMEDOMAIN_VISUALIZATION_DATA
+
+
+
+
var HF_TUNING_INDICATOR_DATA
+
+
+
+
var IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT
+
+
+
+
var IF_DATA_16BIT_REAL_REAL
+
+
+
+
var IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT
+
+
+
+
var IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED
+
+
+
+
var IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED
+
+
+
+
var IMAGE_DATA
+
+
+
+
var INSTANTANEOUS_DATA
+
+
+
+
var LEVEL_DATA
+
+
+
+
var PULSE_DESCRIPTION_WORD_DATA
+
+
+
+
var PULSE_REPETITION_WORD_DATA
+
+
+
+
var SCAN_LEVEL
+
+
+
+
var SCAN_LEVEL_TUNING
+
+
+
+
var SCAN_TUNING
+
+
+
+
var SEGMENTATION_SPECTRUM_DATA_32BIT_FLOATINGPOINT
+
+
+
+
var SPECTRUM_DATA_16BIT
+
+
+
+
var SPECTRUM_DATA_32BIT
+
+
+
+
var SPECTRUM_DATA_32BIT_FLOATINGPOINT
+
+
+
+
var SPECTRUM_DATA_8BIT
+
+
+
+
var SYMBOL_DATA
+
+
+
+
var TEST_DATA
+
+
+
+
var TRANSMISSION_SYSTEM_RESULT_DATA
+
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosContainer.html =================================================================== --- doc/ammosreader/AmmosContainer.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosContainer.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,354 @@ + + + + + + +ammosreader.AmmosContainer API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosContainer

+
+
+

I store the content of an R&S Ammos file in a more accessible way.

+
+ +Expand source code + +
"""I store the content of an R&S Ammos file in a more accessible way."""
+
+class AmmosContainer():
+
+    def __init__(self, name, frames):
+        self.__name = name
+        self.__global_frames = frames
+        self.__tags = []
+
+    @property
+    def name(self):
+        return self.__name
+
+    @name.setter
+    def name(self, a_name):
+        self.__name = a_name
+
+    @property
+    def global_frames(self):
+        return self.__global_frames
+
+    def add_tag(self, tag):
+        self.__tags.append(tag)
+
+    def add_frame(self, frame):
+        self.__global_frames.append(frame)
+
+    def size(self):
+        return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+
+    def frequencies(self):
+        """I return a list of unique frequencies inside this container."""
+        return list({each.global_frame_body.data_header.frequency for each in self.__global_frames})
+
+    def frame_types(self):
+        return list({each.global_frame_header.frame_type for each in self.__global_frames})
+
+    def frame_sizes(self):
+        return [each.global_frame_header.frame_length for each in self.__global_frames]
+
+    def unique_frame_sizes(self):
+        return list({each.global_frame_header.frame_length for each in self.__global_frames})
+
+    def is_homogenic(self):
+        return (len(self.unique_frame_sizes()) == 1) and (len(self.frame_types()) == 1)
+
+    def __str__(self):
+        start_time = self.__global_frames[0].global_frame_body.data_header.timestamp
+        end_time = self.__global_frames[-1].global_frame_body.data_header.timestamp
+
+        frq = str(self.__global_frames[0].global_frame_body.data_header.frequency)
+
+        return ("Start time: " + str(start_time) +
+                "\nEnd time  : " + str(end_time) + "\nFrequencies: " + frq)
+
+    def data_only(self):
+        return [each.global_frame_body.data_body.data for each in self.__global_frames]
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosContainer +(name, frames) +
+
+
+
+ +Expand source code + +
class AmmosContainer():
+
+    def __init__(self, name, frames):
+        self.__name = name
+        self.__global_frames = frames
+        self.__tags = []
+
+    @property
+    def name(self):
+        return self.__name
+
+    @name.setter
+    def name(self, a_name):
+        self.__name = a_name
+
+    @property
+    def global_frames(self):
+        return self.__global_frames
+
+    def add_tag(self, tag):
+        self.__tags.append(tag)
+
+    def add_frame(self, frame):
+        self.__global_frames.append(frame)
+
+    def size(self):
+        return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+
+    def frequencies(self):
+        """I return a list of unique frequencies inside this container."""
+        return list({each.global_frame_body.data_header.frequency for each in self.__global_frames})
+
+    def frame_types(self):
+        return list({each.global_frame_header.frame_type for each in self.__global_frames})
+
+    def frame_sizes(self):
+        return [each.global_frame_header.frame_length for each in self.__global_frames]
+
+    def unique_frame_sizes(self):
+        return list({each.global_frame_header.frame_length for each in self.__global_frames})
+
+    def is_homogenic(self):
+        return (len(self.unique_frame_sizes()) == 1) and (len(self.frame_types()) == 1)
+
+    def __str__(self):
+        start_time = self.__global_frames[0].global_frame_body.data_header.timestamp
+        end_time = self.__global_frames[-1].global_frame_body.data_header.timestamp
+
+        frq = str(self.__global_frames[0].global_frame_body.data_header.frequency)
+
+        return ("Start time: " + str(start_time) +
+                "\nEnd time  : " + str(end_time) + "\nFrequencies: " + frq)
+
+    def data_only(self):
+        return [each.global_frame_body.data_body.data for each in self.__global_frames]
+
+

Instance variables

+
+
var global_frames
+
+
+
+ +Expand source code + +
@property
+def global_frames(self):
+    return self.__global_frames
+
+
+
var name
+
+
+
+ +Expand source code + +
@property
+def name(self):
+    return self.__name
+
+
+
+

Methods

+
+
+def add_frame(self, frame) +
+
+
+
+ +Expand source code + +
def add_frame(self, frame):
+    self.__global_frames.append(frame)
+
+
+
+def add_tag(self, tag) +
+
+
+
+ +Expand source code + +
def add_tag(self, tag):
+    self.__tags.append(tag)
+
+
+
+def data_only(self) +
+
+
+
+ +Expand source code + +
def data_only(self):
+    return [each.global_frame_body.data_body.data for each in self.__global_frames]
+
+
+
+def frame_sizes(self) +
+
+
+
+ +Expand source code + +
def frame_sizes(self):
+    return [each.global_frame_header.frame_length for each in self.__global_frames]
+
+
+
+def frame_types(self) +
+
+
+
+ +Expand source code + +
def frame_types(self):
+    return list({each.global_frame_header.frame_type for each in self.__global_frames})
+
+
+
+def frequencies(self) +
+
+

I return a list of unique frequencies inside this container.

+
+ +Expand source code + +
def frequencies(self):
+    """I return a list of unique frequencies inside this container."""
+    return list({each.global_frame_body.data_header.frequency for each in self.__global_frames})
+
+
+
+def is_homogenic(self) +
+
+
+
+ +Expand source code + +
def is_homogenic(self):
+    return (len(self.unique_frame_sizes()) == 1) and (len(self.frame_types()) == 1)
+
+
+
+def size(self) +
+
+
+
+ +Expand source code + +
def size(self):
+    return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+
+
+
+def unique_frame_sizes(self) +
+
+
+
+ +Expand source code + +
def unique_frame_sizes(self):
+    return list({each.global_frame_header.frame_length for each in self.__global_frames})
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosExtendedAudioDataHeader.html =================================================================== --- doc/ammosreader/AmmosExtendedAudioDataHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosExtendedAudioDataHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,197 @@ + + + + + + +ammosreader.AmmosExtendedAudioDataHeader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosExtendedAudioDataHeader

+
+
+

I provide an Ammos extended data header for audio data frames.

+
+ +Expand source code + +
"""I provide an Ammos extended data header for audio data frames."""
+
+import struct
+from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
+
+
+class AmmosExtendedAudioDataHeader:
+
+    HEADER_SIZE = 44  # 11 words
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return a new AMMOS extended data header for audio frames built from given bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+        standard_header = AmmosAudioDataHeader.from_bytes(bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
+        extended_header_elements = struct.unpack('<Q', bytes[AmmosAudioDataHeader.HEADER_SIZE:])
+        timestamp = extended_header_elements[0]
+        sample_rate = standard_header.sample_rate
+        status = standard_header.status
+        frequency = standard_header.frequency
+        demod_bandwidth = standard_header.demod_bandwidth
+        demod_type = standard_header.demod_type
+        number_of_samples = standard_header.number_of_samples
+        number_of_channels = standard_header.number_of_channels
+        sample_size = standard_header.sample_size
+        return AmmosExtendedAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
+                                            number_of_samples, number_of_channels, sample_size, timestamp)
+
+    def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type,
+                 number_of_samples, number_of_channels, sample_size, timestamp):
+        """I return a new AMMOS extended data header for audio frames built from given parameters."""
+        self.sample_rate = sample_rate
+        self.status = status
+        self.frequency = frequency
+        self.demod_bandwidth = demod_bandwidth
+        self.demod_type = demod_type
+        self.number_of_samples = number_of_samples
+        self.number_of_channels = number_of_channels
+        self.sample_size = sample_size
+        self.timestamp = timestamp
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosExtendedAudioDataHeader +(sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples, number_of_channels, sample_size, timestamp) +
+
+

I return a new AMMOS extended data header for audio frames built from given parameters.

+
+ +Expand source code + +
class AmmosExtendedAudioDataHeader:
+
+    HEADER_SIZE = 44  # 11 words
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return a new AMMOS extended data header for audio frames built from given bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+        standard_header = AmmosAudioDataHeader.from_bytes(bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
+        extended_header_elements = struct.unpack('<Q', bytes[AmmosAudioDataHeader.HEADER_SIZE:])
+        timestamp = extended_header_elements[0]
+        sample_rate = standard_header.sample_rate
+        status = standard_header.status
+        frequency = standard_header.frequency
+        demod_bandwidth = standard_header.demod_bandwidth
+        demod_type = standard_header.demod_type
+        number_of_samples = standard_header.number_of_samples
+        number_of_channels = standard_header.number_of_channels
+        sample_size = standard_header.sample_size
+        return AmmosExtendedAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
+                                            number_of_samples, number_of_channels, sample_size, timestamp)
+
+    def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type,
+                 number_of_samples, number_of_channels, sample_size, timestamp):
+        """I return a new AMMOS extended data header for audio frames built from given parameters."""
+        self.sample_rate = sample_rate
+        self.status = status
+        self.frequency = frequency
+        self.demod_bandwidth = demod_bandwidth
+        self.demod_type = demod_type
+        self.number_of_samples = number_of_samples
+        self.number_of_channels = number_of_channels
+        self.sample_size = sample_size
+        self.timestamp = timestamp
+
+

Class variables

+
+
var HEADER_SIZE
+
+
+
+
+

Static methods

+
+
+def from_bytes(bytes) +
+
+

I return a new AMMOS extended data header for audio frames built from given bytes.

+
+ +Expand source code + +
@classmethod
+def from_bytes(cls, bytes):
+    """I return a new AMMOS extended data header for audio frames built from given bytes."""
+    assert len(bytes) == cls.HEADER_SIZE
+    standard_header = AmmosAudioDataHeader.from_bytes(bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
+    extended_header_elements = struct.unpack('<Q', bytes[AmmosAudioDataHeader.HEADER_SIZE:])
+    timestamp = extended_header_elements[0]
+    sample_rate = standard_header.sample_rate
+    status = standard_header.status
+    frequency = standard_header.frequency
+    demod_bandwidth = standard_header.demod_bandwidth
+    demod_type = standard_header.demod_type
+    number_of_samples = standard_header.number_of_samples
+    number_of_channels = standard_header.number_of_channels
+    sample_size = standard_header.sample_size
+    return AmmosExtendedAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
+                                        number_of_samples, number_of_channels, sample_size, timestamp)
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosExtendedIFDataHeader.html =================================================================== --- doc/ammosreader/AmmosExtendedIFDataHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosExtendedIFDataHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,268 @@ + + + + + + +ammosreader.AmmosExtendedIFDataHeader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosExtendedIFDataHeader

+
+
+
+ +Expand source code + +
import struct
+import numpy as np
+from ammosreader.AmmosIFDataHeader import AmmosIFDataHeader
+
+
+class AmmosExtendedIFDataHeader():
+    """I implement an Ammos extended data header for IF data frames."""
+
+    HEADER_SIZE = 76
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return an AMMOS extended data header from given bytes."""
+        standard_header = AmmosIFDataHeader.from_bytes(bytes[0:AmmosIFDataHeader.HEADER_SIZE])
+        extended_header_elements = struct.unpack('<QQI', bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
+        block_count = standard_header.block_count
+        block_length = standard_header.block_length
+        timestamp = standard_header.timestamp
+        status = standard_header.status
+        source_id = standard_header.source_id
+        source_state = standard_header.source_state
+        frequency = standard_header.frequency
+        bandwidth = standard_header.bandwidth
+        sample_rate = standard_header.sample_rate
+        interpolation = standard_header.interpolation
+        decimation = standard_header.decimation
+        voltage_ref = standard_header.voltage_ref
+        stream_start = np.datetime64(int(extended_header_elements[0]), 'ns')
+        sample_counter = extended_header_elements[1]
+        antenna_correction = extended_header_elements[2]
+        size = len(bytes)
+        return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
+                                         source_state, frequency, bandwidth, sample_rate, interpolation,
+                                         decimation, voltage_ref, stream_start, sample_counter,
+                                         antenna_correction)
+
+    def __init__(self, size, block_count, block_length, timestamp, status, source_id, source_state, frequency,
+                 bandwidth, sample_rate, interpolation, decimation, voltage_ref, stream_start, sample_counter,
+                 antenna_correction):
+        """I create a new instance of myself using the above parameters."""
+        self.size = size
+        self.block_count = block_count
+        self.block_length = block_length
+        self.timestamp = timestamp
+        self.status = status
+        self.source_id = source_id
+        self.source_state = source_state
+        self.frequency = frequency
+        self.bandwidth = bandwidth
+        self.sample_rate = sample_rate
+        self.interpolation = interpolation
+        self.decimation = decimation
+        self.voltage_ref = voltage_ref
+        self.stream_start = stream_start
+        self.sample_counter = sample_counter
+        self.antenna_correction = antenna_correction
+
+    def __str__(self):
+        output = (
+            "\nGlobal frame body data header\n" +
+            "-----------------------------\n" +
+            "Block count:" + str(self.block_count) + "\n" +
+            "Block length:" + str(self.block_length) + "\n" +
+            "Time stamp:" + str(self.timestamp) + "\n" +
+            "Frequency:" + str(self.frequency) + "\n" +
+            "Bandwidth:" + str(self.bandwidth) + "\n" +
+            "Sample rate:" + str(self.sample_rate) + "\n" +
+            "Stream start:" + str(self.stream_start) + "\n" +
+            "Sample counter:" + str(self.sample_counter) + "\n" +
+            "Antenna correction:" + str(self.antenna_correction) + "\n")
+        return output
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosExtendedIFDataHeader +(size, block_count, block_length, timestamp, status, source_id, source_state, frequency, bandwidth, sample_rate, interpolation, decimation, voltage_ref, stream_start, sample_counter, antenna_correction) +
+
+

I implement an Ammos extended data header for IF data frames.

+

I create a new instance of myself using the above parameters.

+
+ +Expand source code + +
class AmmosExtendedIFDataHeader():
+    """I implement an Ammos extended data header for IF data frames."""
+
+    HEADER_SIZE = 76
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return an AMMOS extended data header from given bytes."""
+        standard_header = AmmosIFDataHeader.from_bytes(bytes[0:AmmosIFDataHeader.HEADER_SIZE])
+        extended_header_elements = struct.unpack('<QQI', bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
+        block_count = standard_header.block_count
+        block_length = standard_header.block_length
+        timestamp = standard_header.timestamp
+        status = standard_header.status
+        source_id = standard_header.source_id
+        source_state = standard_header.source_state
+        frequency = standard_header.frequency
+        bandwidth = standard_header.bandwidth
+        sample_rate = standard_header.sample_rate
+        interpolation = standard_header.interpolation
+        decimation = standard_header.decimation
+        voltage_ref = standard_header.voltage_ref
+        stream_start = np.datetime64(int(extended_header_elements[0]), 'ns')
+        sample_counter = extended_header_elements[1]
+        antenna_correction = extended_header_elements[2]
+        size = len(bytes)
+        return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
+                                         source_state, frequency, bandwidth, sample_rate, interpolation,
+                                         decimation, voltage_ref, stream_start, sample_counter,
+                                         antenna_correction)
+
+    def __init__(self, size, block_count, block_length, timestamp, status, source_id, source_state, frequency,
+                 bandwidth, sample_rate, interpolation, decimation, voltage_ref, stream_start, sample_counter,
+                 antenna_correction):
+        """I create a new instance of myself using the above parameters."""
+        self.size = size
+        self.block_count = block_count
+        self.block_length = block_length
+        self.timestamp = timestamp
+        self.status = status
+        self.source_id = source_id
+        self.source_state = source_state
+        self.frequency = frequency
+        self.bandwidth = bandwidth
+        self.sample_rate = sample_rate
+        self.interpolation = interpolation
+        self.decimation = decimation
+        self.voltage_ref = voltage_ref
+        self.stream_start = stream_start
+        self.sample_counter = sample_counter
+        self.antenna_correction = antenna_correction
+
+    def __str__(self):
+        output = (
+            "\nGlobal frame body data header\n" +
+            "-----------------------------\n" +
+            "Block count:" + str(self.block_count) + "\n" +
+            "Block length:" + str(self.block_length) + "\n" +
+            "Time stamp:" + str(self.timestamp) + "\n" +
+            "Frequency:" + str(self.frequency) + "\n" +
+            "Bandwidth:" + str(self.bandwidth) + "\n" +
+            "Sample rate:" + str(self.sample_rate) + "\n" +
+            "Stream start:" + str(self.stream_start) + "\n" +
+            "Sample counter:" + str(self.sample_counter) + "\n" +
+            "Antenna correction:" + str(self.antenna_correction) + "\n")
+        return output
+
+

Class variables

+
+
var HEADER_SIZE
+
+
+
+
+

Static methods

+
+
+def from_bytes(bytes) +
+
+

I return an AMMOS extended data header from given bytes.

+
+ +Expand source code + +
@classmethod
+def from_bytes(cls, bytes):
+    """I return an AMMOS extended data header from given bytes."""
+    standard_header = AmmosIFDataHeader.from_bytes(bytes[0:AmmosIFDataHeader.HEADER_SIZE])
+    extended_header_elements = struct.unpack('<QQI', bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
+    block_count = standard_header.block_count
+    block_length = standard_header.block_length
+    timestamp = standard_header.timestamp
+    status = standard_header.status
+    source_id = standard_header.source_id
+    source_state = standard_header.source_state
+    frequency = standard_header.frequency
+    bandwidth = standard_header.bandwidth
+    sample_rate = standard_header.sample_rate
+    interpolation = standard_header.interpolation
+    decimation = standard_header.decimation
+    voltage_ref = standard_header.voltage_ref
+    stream_start = np.datetime64(int(extended_header_elements[0]), 'ns')
+    sample_counter = extended_header_elements[1]
+    antenna_correction = extended_header_elements[2]
+    size = len(bytes)
+    return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
+                                     source_state, frequency, bandwidth, sample_rate, interpolation,
+                                     decimation, voltage_ref, stream_start, sample_counter,
+                                     antenna_correction)
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosGlobalFrameBody.html =================================================================== --- doc/ammosreader/AmmosGlobalFrameBody.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosGlobalFrameBody.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,197 @@ + + + + + + +ammosreader.AmmosGlobalFrameBody API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosGlobalFrameBody

+
+
+

I provide an AMMOS global frame body.

+
+ +Expand source code + +
"""I provide an AMMOS global frame body."""
+
+class AmmosGlobalFrameBody():
+    """
+    I implement an AMMOS global frame body.
+
+    AMMOS frames can store data of various types.
+    The data is described in the data header.
+    The raw data is then stored in the data body.
+    """
+    def __init__(self, data_header, data_body):
+        self.__data_header = data_header
+        self.__data_body = data_body
+
+    @property
+    def data_header(self):
+        return self.__data_header
+
+    @data_header.setter
+    def data_header(self, a_data_header):
+        self.__data_header = a_data_header
+
+    @property
+    def data_body(self):
+        return self.__data_body
+
+    @data_body.setter
+    def data_body(self, data_bytes):
+        self.__data_body = data_bytes
+
+    def payload(self):
+        """I return the payload only."""
+        return b"".join([each_block.data for each_block in self.data_body])
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosGlobalFrameBody +(data_header, data_body) +
+
+

I implement an AMMOS global frame body.

+

AMMOS frames can store data of various types. +The data is described in the data header. +The raw data is then stored in the data body.

+
+ +Expand source code + +
class AmmosGlobalFrameBody():
+    """
+    I implement an AMMOS global frame body.
+
+    AMMOS frames can store data of various types.
+    The data is described in the data header.
+    The raw data is then stored in the data body.
+    """
+    def __init__(self, data_header, data_body):
+        self.__data_header = data_header
+        self.__data_body = data_body
+
+    @property
+    def data_header(self):
+        return self.__data_header
+
+    @data_header.setter
+    def data_header(self, a_data_header):
+        self.__data_header = a_data_header
+
+    @property
+    def data_body(self):
+        return self.__data_body
+
+    @data_body.setter
+    def data_body(self, data_bytes):
+        self.__data_body = data_bytes
+
+    def payload(self):
+        """I return the payload only."""
+        return b"".join([each_block.data for each_block in self.data_body])
+
+

Instance variables

+
+
var data_body
+
+
+
+ +Expand source code + +
@property
+def data_body(self):
+    return self.__data_body
+
+
+
var data_header
+
+
+
+ +Expand source code + +
@property
+def data_header(self):
+    return self.__data_header
+
+
+
+

Methods

+
+
+def payload(self) +
+
+

I return the payload only.

+
+ +Expand source code + +
def payload(self):
+    """I return the payload only."""
+    return b"".join([each_block.data for each_block in self.data_body])
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosGlobalFrameHeader.html =================================================================== --- doc/ammosreader/AmmosGlobalFrameHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosGlobalFrameHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,251 @@ + + + + + + +ammosreader.AmmosGlobalFrameHeader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosGlobalFrameHeader

+
+
+

I provide an AMMOS global frame header.

+
+ +Expand source code + +
"""I provide an AMMOS global frame header."""
+import struct
+import logging
+
+logging.basicConfig(filename='ammos.log', level=logging.DEBUG)
+
+
+class AmmosGlobalFrameHeader:
+    """I implement an AMMOS global frame header."""
+
+    MAGIC_WORD = "726574fb"
+    HEADER_SIZE = 24
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I create a new AmmosGlobalFrameHeader from bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+
+        elements = struct.unpack('<4s4s4s4s4s4s', bytes)
+
+        magic_word = elements[0].hex()
+
+        if magic_word != cls.MAGIC_WORD:
+            return None
+
+        frame_length = (int.from_bytes(elements[1], byteorder='little')*4)
+
+        running_frame_number = (int.from_bytes(elements[2], byteorder='little'))
+
+        frame_type = (int.from_bytes(elements[3], byteorder='little'))
+
+        data_header_length = 4 * int.from_bytes(elements[4], byteorder='little')
+
+        reserved = elements[5]
+
+        return AmmosGlobalFrameHeader(magic_word, frame_length, running_frame_number,
+                                      frame_type, data_header_length, reserved)
+
+    def __init__(self, magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved):
+        """I return a new instance of myself initialized with above parameters."""
+        if magic_word != type(self).MAGIC_WORD:
+            logging.error("Wrong magic word found")
+            self.magic_word = magic_word
+        else:
+            self.magic_word = magic_word
+            self.frame_length = frame_length
+            self.running_frame_number = running_frame_number
+            self.frame_type = frame_type
+            self.data_header_length = data_header_length
+            self.reserved = reserved
+
+    def __str__(self):
+        """I return the string representation of myself."""
+        output = ("Global frame header info\n" +
+                  "------------------------\n" +
+                  "Magic word:" + str(self.magic_word) + "\n" +
+                  "Frame length:" + str(self.frame_length) + "\n" +
+                  "Running frame:" + str(self.running_frame_number) + "\n" +
+                  "Frame Type:" + str(self.frame_type) + "\n" +
+                  "Data header length:" + str(self.data_header_length) + "\n")
+        return output
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosGlobalFrameHeader +(magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved) +
+
+

I implement an AMMOS global frame header.

+

I return a new instance of myself initialized with above parameters.

+
+ +Expand source code + +
class AmmosGlobalFrameHeader:
+    """I implement an AMMOS global frame header."""
+
+    MAGIC_WORD = "726574fb"
+    HEADER_SIZE = 24
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I create a new AmmosGlobalFrameHeader from bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+
+        elements = struct.unpack('<4s4s4s4s4s4s', bytes)
+
+        magic_word = elements[0].hex()
+
+        if magic_word != cls.MAGIC_WORD:
+            return None
+
+        frame_length = (int.from_bytes(elements[1], byteorder='little')*4)
+
+        running_frame_number = (int.from_bytes(elements[2], byteorder='little'))
+
+        frame_type = (int.from_bytes(elements[3], byteorder='little'))
+
+        data_header_length = 4 * int.from_bytes(elements[4], byteorder='little')
+
+        reserved = elements[5]
+
+        return AmmosGlobalFrameHeader(magic_word, frame_length, running_frame_number,
+                                      frame_type, data_header_length, reserved)
+
+    def __init__(self, magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved):
+        """I return a new instance of myself initialized with above parameters."""
+        if magic_word != type(self).MAGIC_WORD:
+            logging.error("Wrong magic word found")
+            self.magic_word = magic_word
+        else:
+            self.magic_word = magic_word
+            self.frame_length = frame_length
+            self.running_frame_number = running_frame_number
+            self.frame_type = frame_type
+            self.data_header_length = data_header_length
+            self.reserved = reserved
+
+    def __str__(self):
+        """I return the string representation of myself."""
+        output = ("Global frame header info\n" +
+                  "------------------------\n" +
+                  "Magic word:" + str(self.magic_word) + "\n" +
+                  "Frame length:" + str(self.frame_length) + "\n" +
+                  "Running frame:" + str(self.running_frame_number) + "\n" +
+                  "Frame Type:" + str(self.frame_type) + "\n" +
+                  "Data header length:" + str(self.data_header_length) + "\n")
+        return output
+
+

Class variables

+
+
var HEADER_SIZE
+
+
+
+
var MAGIC_WORD
+
+
+
+
+

Static methods

+
+
+def from_bytes(bytes) +
+
+

I create a new AmmosGlobalFrameHeader from bytes.

+
+ +Expand source code + +
@classmethod
+def from_bytes(cls, bytes):
+    """I create a new AmmosGlobalFrameHeader from bytes."""
+    assert len(bytes) == cls.HEADER_SIZE
+
+    elements = struct.unpack('<4s4s4s4s4s4s', bytes)
+
+    magic_word = elements[0].hex()
+
+    if magic_word != cls.MAGIC_WORD:
+        return None
+
+    frame_length = (int.from_bytes(elements[1], byteorder='little')*4)
+
+    running_frame_number = (int.from_bytes(elements[2], byteorder='little'))
+
+    frame_type = (int.from_bytes(elements[3], byteorder='little'))
+
+    data_header_length = 4 * int.from_bytes(elements[4], byteorder='little')
+
+    reserved = elements[5]
+
+    return AmmosGlobalFrameHeader(magic_word, frame_length, running_frame_number,
+                                  frame_type, data_header_length, reserved)
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosIFDataBlock.html =================================================================== --- doc/ammosreader/AmmosIFDataBlock.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosIFDataBlock.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,137 @@ + + + + + + +ammosreader.AmmosIFDataBlock API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosIFDataBlock

+
+
+
+ +Expand source code + +
class AmmosIFDataBlock:
+
+    def __init__(self, header, data):
+        self.__header = header
+        self.__data = data
+
+    @property
+    def header(self):
+        return self.__header
+
+    @property
+    def payload(self):
+        return self.__data
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosIFDataBlock +(header, data) +
+
+
+
+ +Expand source code + +
class AmmosIFDataBlock:
+
+    def __init__(self, header, data):
+        self.__header = header
+        self.__data = data
+
+    @property
+    def header(self):
+        return self.__header
+
+    @property
+    def payload(self):
+        return self.__data
+
+

Instance variables

+
+
var header
+
+
+
+ +Expand source code + +
@property
+def header(self):
+    return self.__header
+
+
+
var payload
+
+
+
+ +Expand source code + +
@property
+def payload(self):
+    return self.__data
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosIFDataBlockHeader.html =================================================================== --- doc/ammosreader/AmmosIFDataBlockHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosIFDataBlockHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,325 @@ + + + + + + +ammosreader.AmmosIFDataBlockHeader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosIFDataBlockHeader

+
+
+
+ +Expand source code + +
import struct
+
+
+class AmmosIFDataBlockHeader:
+    """I implement an AMMOS IF data block header."""
+
+    HEADER_SIZE = 4
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return an AMMOS IF data block header built from bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+        elements = struct.unpack('<ccH', bytes)
+        header = cls()
+        first_entry = int.from_bytes(elements[0], byteorder='little')
+        header.invalidity = ((first_entry & 1) == 1)
+        header.blanking = ((first_entry & 2) == 1)
+        header.user_data = ((first_entry & 252))
+        header.reserved = int.from_bytes(elements[1], byteorder='little')
+        header.reciprocal_gain = elements[2]
+        return header
+
+    def __init__(self):
+        """I return an AMMOS IF data block header initialized with sane default values."""
+        self.__invalidity = True
+        self.__blanking = False
+        self.__user_data = 0
+        self.__reserved = 0
+        self.__reciprocal_gain = 0
+
+    @property
+    def invalidity(self):
+        return self.__invalidity
+
+    @invalidity.setter
+    def invalidity(self, a_boolean):
+        self.__invalidity = a_boolean
+
+    @property
+    def blanking(self):
+        return self.__blanking
+
+    @blanking.setter
+    def blanking(self, a_boolean):
+        self.__blanking = a_boolean
+
+    @property
+    def user_data(self):
+        return self.__user_data
+
+    @user_data.setter
+    def user_data(self, some_data):
+        self.__user_data = some_data
+
+    @property
+    def reserved(self):
+        return self.__reserved
+
+    @reserved.setter
+    def reserved(self, some_data):
+        assert some_data == 0
+        self.__reserved = some_data
+
+    @property
+    def reciprocal_gain(self):
+        return self.__reciprocal_gain
+
+    @reciprocal_gain.setter
+    def reciprocal_gain(self, a_gain_value):
+        assert 0 <= a_gain_value < pow(2, 16)
+        self.__reciprocal_gain = a_gain_value
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosIFDataBlockHeader +
+
+

I implement an AMMOS IF data block header.

+

I return an AMMOS IF data block header initialized with sane default values.

+
+ +Expand source code + +
class AmmosIFDataBlockHeader:
+    """I implement an AMMOS IF data block header."""
+
+    HEADER_SIZE = 4
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return an AMMOS IF data block header built from bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+        elements = struct.unpack('<ccH', bytes)
+        header = cls()
+        first_entry = int.from_bytes(elements[0], byteorder='little')
+        header.invalidity = ((first_entry & 1) == 1)
+        header.blanking = ((first_entry & 2) == 1)
+        header.user_data = ((first_entry & 252))
+        header.reserved = int.from_bytes(elements[1], byteorder='little')
+        header.reciprocal_gain = elements[2]
+        return header
+
+    def __init__(self):
+        """I return an AMMOS IF data block header initialized with sane default values."""
+        self.__invalidity = True
+        self.__blanking = False
+        self.__user_data = 0
+        self.__reserved = 0
+        self.__reciprocal_gain = 0
+
+    @property
+    def invalidity(self):
+        return self.__invalidity
+
+    @invalidity.setter
+    def invalidity(self, a_boolean):
+        self.__invalidity = a_boolean
+
+    @property
+    def blanking(self):
+        return self.__blanking
+
+    @blanking.setter
+    def blanking(self, a_boolean):
+        self.__blanking = a_boolean
+
+    @property
+    def user_data(self):
+        return self.__user_data
+
+    @user_data.setter
+    def user_data(self, some_data):
+        self.__user_data = some_data
+
+    @property
+    def reserved(self):
+        return self.__reserved
+
+    @reserved.setter
+    def reserved(self, some_data):
+        assert some_data == 0
+        self.__reserved = some_data
+
+    @property
+    def reciprocal_gain(self):
+        return self.__reciprocal_gain
+
+    @reciprocal_gain.setter
+    def reciprocal_gain(self, a_gain_value):
+        assert 0 <= a_gain_value < pow(2, 16)
+        self.__reciprocal_gain = a_gain_value
+
+

Class variables

+
+
var HEADER_SIZE
+
+
+
+
+

Static methods

+
+
+def from_bytes(bytes) +
+
+

I return an AMMOS IF data block header built from bytes.

+
+ +Expand source code + +
@classmethod
+def from_bytes(cls, bytes):
+    """I return an AMMOS IF data block header built from bytes."""
+    assert len(bytes) == cls.HEADER_SIZE
+    elements = struct.unpack('<ccH', bytes)
+    header = cls()
+    first_entry = int.from_bytes(elements[0], byteorder='little')
+    header.invalidity = ((first_entry & 1) == 1)
+    header.blanking = ((first_entry & 2) == 1)
+    header.user_data = ((first_entry & 252))
+    header.reserved = int.from_bytes(elements[1], byteorder='little')
+    header.reciprocal_gain = elements[2]
+    return header
+
+
+
+

Instance variables

+
+
var blanking
+
+
+
+ +Expand source code + +
@property
+def blanking(self):
+    return self.__blanking
+
+
+
var invalidity
+
+
+
+ +Expand source code + +
@property
+def invalidity(self):
+    return self.__invalidity
+
+
+
var reciprocal_gain
+
+
+
+ +Expand source code + +
@property
+def reciprocal_gain(self):
+    return self.__reciprocal_gain
+
+
+
var reserved
+
+
+
+ +Expand source code + +
@property
+def reserved(self):
+    return self.__reserved
+
+
+
var user_data
+
+
+
+ +Expand source code + +
@property
+def user_data(self):
+    return self.__user_data
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosIFDataBody.html =================================================================== --- doc/ammosreader/AmmosIFDataBody.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosIFDataBody.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,172 @@ + + + + + + +ammosreader.AmmosIFDataBody API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosIFDataBody

+
+
+

I provide an AMMOS data body for IF data.

+

I manage multiple IF data blocks

+
+ +Expand source code + +
"""
+I provide an AMMOS data body for IF data.
+
+I manage multiple IF data blocks
+"""
+
+class AmmosIFDataBody:
+    """I implement an AMMOS data body for IF data."""
+    def __init__(self, data_blocks=None):
+        if data_blocks is None:
+            self.__data_blocks = []
+        else:
+            self.__data_blocks = data_blocks
+
+    @property
+    def data_blocks(self):
+        return self.__data_blocks
+
+    def add_data_block(self, a_data_block):
+        self.__data_blocks.append(a_data_block)
+
+    @property
+    def payload(self):
+        return b"".join([each.payload for each in self.data_blocks])
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosIFDataBody +(data_blocks=None) +
+
+

I implement an AMMOS data body for IF data.

+
+ +Expand source code + +
class AmmosIFDataBody:
+    """I implement an AMMOS data body for IF data."""
+    def __init__(self, data_blocks=None):
+        if data_blocks is None:
+            self.__data_blocks = []
+        else:
+            self.__data_blocks = data_blocks
+
+    @property
+    def data_blocks(self):
+        return self.__data_blocks
+
+    def add_data_block(self, a_data_block):
+        self.__data_blocks.append(a_data_block)
+
+    @property
+    def payload(self):
+        return b"".join([each.payload for each in self.data_blocks])
+
+

Instance variables

+
+
var data_blocks
+
+
+
+ +Expand source code + +
@property
+def data_blocks(self):
+    return self.__data_blocks
+
+
+
var payload
+
+
+
+ +Expand source code + +
@property
+def payload(self):
+    return b"".join([each.payload for each in self.data_blocks])
+
+
+
+

Methods

+
+
+def add_data_block(self, a_data_block) +
+
+
+
+ +Expand source code + +
def add_data_block(self, a_data_block):
+    self.__data_blocks.append(a_data_block)
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosIFDataHeader.html =================================================================== --- doc/ammosreader/AmmosIFDataHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosIFDataHeader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,240 @@ + + + + + + +ammosreader.AmmosIFDataHeader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosIFDataHeader

+
+
+

I provide a Ammos data header for IF data frames.

+
+ +Expand source code + +
"""I provide a Ammos data header for IF data frames."""
+
+import struct
+import numpy as np
+
+
+class AmmosIFDataHeader():
+    """I implement an Ammos data header for IF data frames."""
+
+    HEADER_SIZE = 56
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return an AMMOS data header from given bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+        elements = struct.unpack('<IIQIIIQIIIIi', bytes)
+        block_count = elements[0]
+        block_length = int(elements[1])*4
+        timestamp = np.datetime64(int(elements[2])*1000, 'ns')
+        status = elements[3]
+        source_id = elements[4]
+        source_state = elements[5]
+        frequency = elements[6]
+        bandwidth = elements[7]
+        sample_rate = elements[8]
+        interpolation = elements[9]
+        decimation = elements[10]
+        voltage_ref = elements[11]
+
+        return AmmosIFDataHeader(block_count, block_length, timestamp, status, source_id,
+                                 source_state, frequency, bandwidth, sample_rate,
+                                 interpolation, decimation, voltage_ref)
+
+    def __init__(self, block_count, block_length, timestamp, status, source_id, source_state, frequency,
+                 bandwidth, sample_rate, interpolation, decimation, voltage_ref):
+        """I create a new instance of myself using the above parameters."""
+        self.block_count = block_count
+        self.block_length = block_length
+        self.timestamp = timestamp
+        self.status = status
+        self.source_id = source_id
+        self.source_state = source_state
+        self.frequency = frequency
+        self.bandwidth = bandwidth
+        self.sample_rate = sample_rate
+        self.interpolation = interpolation
+        self.decimation = decimation
+        self.voltage_ref = voltage_ref
+
+    def __str_(self):
+        output = ("\nGlobal frame body data header\n" +
+                  "-----------------------------\n" +
+                  "Block count:" + str(self.block_count) + "\n" +
+                  "Block length:" + str(self.block_length) + "\n" +
+                  "Time stamp:" + str(self.timestamp) + "\n" +
+                  "Frequency:" + str(self.frequency) + "\n" +
+                  "Bandwidth:" + str(self.bandwidth) + "\n" +
+                  "Sample rate:" + str(self.sample_rate) + "\n")
+        return output
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosIFDataHeader +(block_count, block_length, timestamp, status, source_id, source_state, frequency, bandwidth, sample_rate, interpolation, decimation, voltage_ref) +
+
+

I implement an Ammos data header for IF data frames.

+

I create a new instance of myself using the above parameters.

+
+ +Expand source code + +
class AmmosIFDataHeader():
+    """I implement an Ammos data header for IF data frames."""
+
+    HEADER_SIZE = 56
+
+    @classmethod
+    def from_bytes(cls, bytes):
+        """I return an AMMOS data header from given bytes."""
+        assert len(bytes) == cls.HEADER_SIZE
+        elements = struct.unpack('<IIQIIIQIIIIi', bytes)
+        block_count = elements[0]
+        block_length = int(elements[1])*4
+        timestamp = np.datetime64(int(elements[2])*1000, 'ns')
+        status = elements[3]
+        source_id = elements[4]
+        source_state = elements[5]
+        frequency = elements[6]
+        bandwidth = elements[7]
+        sample_rate = elements[8]
+        interpolation = elements[9]
+        decimation = elements[10]
+        voltage_ref = elements[11]
+
+        return AmmosIFDataHeader(block_count, block_length, timestamp, status, source_id,
+                                 source_state, frequency, bandwidth, sample_rate,
+                                 interpolation, decimation, voltage_ref)
+
+    def __init__(self, block_count, block_length, timestamp, status, source_id, source_state, frequency,
+                 bandwidth, sample_rate, interpolation, decimation, voltage_ref):
+        """I create a new instance of myself using the above parameters."""
+        self.block_count = block_count
+        self.block_length = block_length
+        self.timestamp = timestamp
+        self.status = status
+        self.source_id = source_id
+        self.source_state = source_state
+        self.frequency = frequency
+        self.bandwidth = bandwidth
+        self.sample_rate = sample_rate
+        self.interpolation = interpolation
+        self.decimation = decimation
+        self.voltage_ref = voltage_ref
+
+    def __str_(self):
+        output = ("\nGlobal frame body data header\n" +
+                  "-----------------------------\n" +
+                  "Block count:" + str(self.block_count) + "\n" +
+                  "Block length:" + str(self.block_length) + "\n" +
+                  "Time stamp:" + str(self.timestamp) + "\n" +
+                  "Frequency:" + str(self.frequency) + "\n" +
+                  "Bandwidth:" + str(self.bandwidth) + "\n" +
+                  "Sample rate:" + str(self.sample_rate) + "\n")
+        return output
+
+

Class variables

+
+
var HEADER_SIZE
+
+
+
+
+

Static methods

+
+
+def from_bytes(bytes) +
+
+

I return an AMMOS data header from given bytes.

+
+ +Expand source code + +
@classmethod
+def from_bytes(cls, bytes):
+    """I return an AMMOS data header from given bytes."""
+    assert len(bytes) == cls.HEADER_SIZE
+    elements = struct.unpack('<IIQIIIQIIIIi', bytes)
+    block_count = elements[0]
+    block_length = int(elements[1])*4
+    timestamp = np.datetime64(int(elements[2])*1000, 'ns')
+    status = elements[3]
+    source_id = elements[4]
+    source_state = elements[5]
+    frequency = elements[6]
+    bandwidth = elements[7]
+    sample_rate = elements[8]
+    interpolation = elements[9]
+    decimation = elements[10]
+    voltage_ref = elements[11]
+
+    return AmmosIFDataHeader(block_count, block_length, timestamp, status, source_id,
+                             source_state, frequency, bandwidth, sample_rate,
+                             interpolation, decimation, voltage_ref)
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosIFReader.html =================================================================== --- doc/ammosreader/AmmosIFReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosIFReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,463 @@ + + + + + + +ammosreader.AmmosIFReader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosIFReader

+
+
+

I provide a specialized Ammos Reader for IF data.

+
+ +Expand source code + +
"""I provide a specialized Ammos Reader for IF data."""
+import logging
+
+from ammosreader.AbstractAmmosReader import AbstractAmmosReader
+from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody
+from ammosreader.AmmosIFDataHeader import AmmosIFDataHeader
+from ammosreader.AmmosExtendedIFDataHeader import AmmosExtendedIFDataHeader
+from ammosreader.AmmosIFDataBody import AmmosIFDataBody
+from ammosreader.AmmosIFDataBlock import AmmosIFDataBlock
+from ammosreader.AmmosIFDataBlockHeader import AmmosIFDataBlockHeader
+
+
+class AmmosIFReader(AbstractAmmosReader):
+    """I read the IF data embedded in an R&S AMMOS recording."""
+
+    def __init__(self, file_name):
+        super().__init__(file_name)
+
+    def read_next_global_frame_body_data_header(self):
+
+        header_size = AmmosIFDataHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+
+        logging.info("\nReading global frame body standard data header\n")
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame body data header", header_size)
+            return None
+        return AmmosIFDataHeader.from_bytes(bytes)
+
+    def read_next_global_frame_body_extended_data_header(self):
+        """
+        I return the next global frame body extended data header from current position in file.
+
+        :return: the next Ammos Extended IF data header or None if incomplete
+        :rtype: AmmosExtendedIFDataHeader
+        """
+        header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+
+        logging.info("\nReading global frame body extended data header\n")
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
+            return None
+        return AmmosExtendedIFDataHeader.from_bytes(bytes)
+
+    def read_next_if_data_body(self, number_of_data_blocks, data_length):
+        """
+        I return the next data body read from current position in file.
+
+        :param number_of_data_blocks: the number of data blocks inside the body
+        :type number_of_data_blocks: int
+
+        :param data_length: the length of the raw data inside a single block
+        :type data_length: int
+        """
+        header_size = AmmosIFDataBlockHeader.HEADER_SIZE
+
+        data_body = AmmosIFDataBody()
+
+        block_length = header_size + data_length
+
+        total = number_of_data_blocks*block_length
+
+        byte_string = self.ammos_file.read(block_length)
+
+        if len(byte_string) != total:
+            logging.debug("Can not read all %s bytes of data body", total)
+            return None
+
+        for i in range(0, number_of_data_blocks):
+            result = byte_string[i*block_length:(i*block_length+block_length)]
+            data_body.add_data_block(AmmosIFDataBlock(AmmosIFDataBlockHeader.from_bytes(result[0:header_size]),
+                                                      result[header_size:]))
+
+        return data_body
+
+    def read_next_global_frame_body(self, data_header_length):
+        """
+        I return the next global frame body read from current position in file.
+
+        :param data_header_length: the length of the data header
+        :type data_header_length: int
+        """
+        if_data_header = None
+
+        if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
+            if_data_header = self.read_next_global_frame_body_data_header()
+        if data_header_length == AmmosExtendedIFDataHeader.HEADER_SIZE:
+            if_data_header = self.read_next_global_frame_body_extended_data_header()
+
+        if if_data_header is None:
+            logging.debug("Data header missing")
+            return None
+
+        if_data_body = self.read_next_if_data_body(if_data_header.block_count, if_data_header.block_length)
+
+        if if_data_body is None:
+            logging.debug("Data body missing")
+            return None
+
+        return AmmosGlobalFrameBody(if_data_header, if_data_body)
+
+    def payload(self):
+        return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosIFReader +(file_name) +
+
+

I read the IF data embedded in an R&S AMMOS recording.

+

I am the standard constructor for Ammos Readers.

+

Additional information about the file can be added as key/value pairs in tags

+

:param file_name: The file to read Ammos data from +:type file_name: str

+
+ +Expand source code + +
class AmmosIFReader(AbstractAmmosReader):
+    """I read the IF data embedded in an R&S AMMOS recording."""
+
+    def __init__(self, file_name):
+        super().__init__(file_name)
+
+    def read_next_global_frame_body_data_header(self):
+
+        header_size = AmmosIFDataHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+
+        logging.info("\nReading global frame body standard data header\n")
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame body data header", header_size)
+            return None
+        return AmmosIFDataHeader.from_bytes(bytes)
+
+    def read_next_global_frame_body_extended_data_header(self):
+        """
+        I return the next global frame body extended data header from current position in file.
+
+        :return: the next Ammos Extended IF data header or None if incomplete
+        :rtype: AmmosExtendedIFDataHeader
+        """
+        header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
+
+        bytes = self.ammos_file.read(header_size)
+
+        logging.info("\nReading global frame body extended data header\n")
+        if ((not bytes) or (len(bytes) < header_size)):
+            logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
+            return None
+        return AmmosExtendedIFDataHeader.from_bytes(bytes)
+
+    def read_next_if_data_body(self, number_of_data_blocks, data_length):
+        """
+        I return the next data body read from current position in file.
+
+        :param number_of_data_blocks: the number of data blocks inside the body
+        :type number_of_data_blocks: int
+
+        :param data_length: the length of the raw data inside a single block
+        :type data_length: int
+        """
+        header_size = AmmosIFDataBlockHeader.HEADER_SIZE
+
+        data_body = AmmosIFDataBody()
+
+        block_length = header_size + data_length
+
+        total = number_of_data_blocks*block_length
+
+        byte_string = self.ammos_file.read(block_length)
+
+        if len(byte_string) != total:
+            logging.debug("Can not read all %s bytes of data body", total)
+            return None
+
+        for i in range(0, number_of_data_blocks):
+            result = byte_string[i*block_length:(i*block_length+block_length)]
+            data_body.add_data_block(AmmosIFDataBlock(AmmosIFDataBlockHeader.from_bytes(result[0:header_size]),
+                                                      result[header_size:]))
+
+        return data_body
+
+    def read_next_global_frame_body(self, data_header_length):
+        """
+        I return the next global frame body read from current position in file.
+
+        :param data_header_length: the length of the data header
+        :type data_header_length: int
+        """
+        if_data_header = None
+
+        if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
+            if_data_header = self.read_next_global_frame_body_data_header()
+        if data_header_length == AmmosExtendedIFDataHeader.HEADER_SIZE:
+            if_data_header = self.read_next_global_frame_body_extended_data_header()
+
+        if if_data_header is None:
+            logging.debug("Data header missing")
+            return None
+
+        if_data_body = self.read_next_if_data_body(if_data_header.block_count, if_data_header.block_length)
+
+        if if_data_body is None:
+            logging.debug("Data body missing")
+            return None
+
+        return AmmosGlobalFrameBody(if_data_header, if_data_body)
+
+    def payload(self):
+        return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])
+
+

Ancestors

+ +

Methods

+
+
+def payload(self) +
+
+
+
+ +Expand source code + +
def payload(self):
+    return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])
+
+
+
+def read_next_global_frame_body(self, data_header_length) +
+
+

I return the next global frame body read from current position in file.

+

:param data_header_length: the length of the data header +:type data_header_length: int

+
+ +Expand source code + +
def read_next_global_frame_body(self, data_header_length):
+    """
+    I return the next global frame body read from current position in file.
+
+    :param data_header_length: the length of the data header
+    :type data_header_length: int
+    """
+    if_data_header = None
+
+    if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
+        if_data_header = self.read_next_global_frame_body_data_header()
+    if data_header_length == AmmosExtendedIFDataHeader.HEADER_SIZE:
+        if_data_header = self.read_next_global_frame_body_extended_data_header()
+
+    if if_data_header is None:
+        logging.debug("Data header missing")
+        return None
+
+    if_data_body = self.read_next_if_data_body(if_data_header.block_count, if_data_header.block_length)
+
+    if if_data_body is None:
+        logging.debug("Data body missing")
+        return None
+
+    return AmmosGlobalFrameBody(if_data_header, if_data_body)
+
+
+
+def read_next_global_frame_body_data_header(self) +
+
+
+
+ +Expand source code + +
def read_next_global_frame_body_data_header(self):
+
+    header_size = AmmosIFDataHeader.HEADER_SIZE
+
+    bytes = self.ammos_file.read(header_size)
+
+    logging.info("\nReading global frame body standard data header\n")
+    if ((not bytes) or (len(bytes) < header_size)):
+        logging.debug("Can not read all %s bytes of global frame body data header", header_size)
+        return None
+    return AmmosIFDataHeader.from_bytes(bytes)
+
+
+
+def read_next_global_frame_body_extended_data_header(self) +
+
+

I return the next global frame body extended data header from current position in file.

+

:return: the next Ammos Extended IF data header or None if incomplete +:rtype: AmmosExtendedIFDataHeader

+
+ +Expand source code + +
def read_next_global_frame_body_extended_data_header(self):
+    """
+    I return the next global frame body extended data header from current position in file.
+
+    :return: the next Ammos Extended IF data header or None if incomplete
+    :rtype: AmmosExtendedIFDataHeader
+    """
+    header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
+
+    bytes = self.ammos_file.read(header_size)
+
+    logging.info("\nReading global frame body extended data header\n")
+    if ((not bytes) or (len(bytes) < header_size)):
+        logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
+        return None
+    return AmmosExtendedIFDataHeader.from_bytes(bytes)
+
+
+
+def read_next_if_data_body(self, number_of_data_blocks, data_length) +
+
+

I return the next data body read from current position in file.

+

:param number_of_data_blocks: the number of data blocks inside the body +:type number_of_data_blocks: int

+

:param data_length: the length of the raw data inside a single block +:type data_length: int

+
+ +Expand source code + +
def read_next_if_data_body(self, number_of_data_blocks, data_length):
+    """
+    I return the next data body read from current position in file.
+
+    :param number_of_data_blocks: the number of data blocks inside the body
+    :type number_of_data_blocks: int
+
+    :param data_length: the length of the raw data inside a single block
+    :type data_length: int
+    """
+    header_size = AmmosIFDataBlockHeader.HEADER_SIZE
+
+    data_body = AmmosIFDataBody()
+
+    block_length = header_size + data_length
+
+    total = number_of_data_blocks*block_length
+
+    byte_string = self.ammos_file.read(block_length)
+
+    if len(byte_string) != total:
+        logging.debug("Can not read all %s bytes of data body", total)
+        return None
+
+    for i in range(0, number_of_data_blocks):
+        result = byte_string[i*block_length:(i*block_length+block_length)]
+        data_body.add_data_block(AmmosIFDataBlock(AmmosIFDataBlockHeader.from_bytes(result[0:header_size]),
+                                                  result[header_size:]))
+
+    return data_body
+
+
+
+

Inherited members

+ +
+
+
+
+ +
+ + + Index: doc/ammosreader/AmmosSingleFrame.html =================================================================== --- doc/ammosreader/AmmosSingleFrame.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/AmmosSingleFrame.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,210 @@ + + + + + + +ammosreader.AmmosSingleFrame API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.AmmosSingleFrame

+
+
+
+ +Expand source code + +
class AmmosSingleFrame():
+
+    def __init__(self, global_frame_header, global_frame_body):
+        self.__global_frame_header = global_frame_header
+        self.__global_frame_body = global_frame_body
+
+    @property
+    def global_frame_header(self):
+        return self.__global_frame_header
+
+    @property
+    def global_frame_body(self):
+        return self.__global_frame_body
+
+    def data(self):
+        return self.global_frame_body.data_bytes_only()
+
+    def size_correct(self):
+        return (self.global_frame_header.frame_length == (24 + self.global_frame_header.data_header_length +
+                                                          (self.global_frame_body.data_header.block_count *
+                                                           (self.global_frame_body.data_header.block_length + 4))))
+    # FIXME: Use str method instead
+
+    def __str__(self):
+        output = (
+            "Global frame header\n" +
+            "-------------------\n" +
+            "Frame type:" + str(self.global_frame_header.frame_type) + "\n" +
+            "Frame count:" + str(self.global_frame_header.running_frame_number) + "\n" +
+            "Data header length:" + str(self.global_frame_header.data_header_length) + " bytes\n" +
+            "Frame length:" + str(self.global_frame_header.frame_length) + " bytes\n"
+        )
+        return output
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class AmmosSingleFrame +(global_frame_header, global_frame_body) +
+
+
+
+ +Expand source code + +
class AmmosSingleFrame():
+
+    def __init__(self, global_frame_header, global_frame_body):
+        self.__global_frame_header = global_frame_header
+        self.__global_frame_body = global_frame_body
+
+    @property
+    def global_frame_header(self):
+        return self.__global_frame_header
+
+    @property
+    def global_frame_body(self):
+        return self.__global_frame_body
+
+    def data(self):
+        return self.global_frame_body.data_bytes_only()
+
+    def size_correct(self):
+        return (self.global_frame_header.frame_length == (24 + self.global_frame_header.data_header_length +
+                                                          (self.global_frame_body.data_header.block_count *
+                                                           (self.global_frame_body.data_header.block_length + 4))))
+    # FIXME: Use str method instead
+
+    def __str__(self):
+        output = (
+            "Global frame header\n" +
+            "-------------------\n" +
+            "Frame type:" + str(self.global_frame_header.frame_type) + "\n" +
+            "Frame count:" + str(self.global_frame_header.running_frame_number) + "\n" +
+            "Data header length:" + str(self.global_frame_header.data_header_length) + " bytes\n" +
+            "Frame length:" + str(self.global_frame_header.frame_length) + " bytes\n"
+        )
+        return output
+
+

Instance variables

+
+
var global_frame_body
+
+
+
+ +Expand source code + +
@property
+def global_frame_body(self):
+    return self.__global_frame_body
+
+
+
var global_frame_header
+
+
+
+ +Expand source code + +
@property
+def global_frame_header(self):
+    return self.__global_frame_header
+
+
+
+

Methods

+
+
+def data(self) +
+
+
+
+ +Expand source code + +
def data(self):
+    return self.global_frame_body.data_bytes_only()
+
+
+
+def size_correct(self) +
+
+
+
+ +Expand source code + +
def size_correct(self):
+    return (self.global_frame_header.frame_length == (24 + self.global_frame_header.data_header_length +
+                                                      (self.global_frame_body.data_header.block_count *
+                                                       (self.global_frame_body.data_header.block_length + 4))))
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/IQDWTXBlock.html =================================================================== --- doc/ammosreader/IQDWTXBlock.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/IQDWTXBlock.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,126 @@ + + + + + + +ammosreader.IQDWTXBlock API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.IQDWTXBlock

+
+
+
+ +Expand source code + +
from ammosreader.PDW import PDW
+
+
+class IQDWTXBLOCK():
+
+    @classmethod
+    def from_bytes(cls, byte_string):
+        pdw = PDW.from_bytes(bytes[0:32])
+        return IQDWTXBLOCK(pdw, bytes[32:])
+
+    def __init__(self, pdw, iq):
+
+        self.pdw = pdw
+        self.iq = iq
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class IQDWTXBLOCK +(pdw, iq) +
+
+
+
+ +Expand source code + +
class IQDWTXBLOCK():
+
+    @classmethod
+    def from_bytes(cls, byte_string):
+        pdw = PDW.from_bytes(bytes[0:32])
+        return IQDWTXBLOCK(pdw, bytes[32:])
+
+    def __init__(self, pdw, iq):
+
+        self.pdw = pdw
+        self.iq = iq
+
+

Static methods

+
+
+def from_bytes(byte_string) +
+
+
+
+ +Expand source code + +
@classmethod
+def from_bytes(cls, byte_string):
+    pdw = PDW.from_bytes(bytes[0:32])
+    return IQDWTXBLOCK(pdw, bytes[32:])
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/PDW.html =================================================================== --- doc/ammosreader/PDW.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/PDW.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,756 @@ + + + + + + +ammosreader.PDW API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.PDW

+
+
+

I store the information of a single PDW block.

+
+ +Expand source code + +
"""I store the information of a single PDW block."""
+
+import struct
+import math
+import numpy as np
+
+
+class PDW():
+    """
+    I store information from a single ppdw data block.
+
+    .. automethod:: __init__
+    """
+
+    @classmethod
+    def from_bytes(cls, byte_string):
+        """
+        I create an instance of class PDW from data body (8 * 32 bits).
+
+        :param byte_string: a byte string containing a single data body read from a ppdw file
+        :type byte_string: byte string
+
+        :return: an instance of class PDW with attributes set according to the data of a data body
+        :rtype: PDW
+        """
+        assert(len(byte_string) == 32)
+
+        parts = struct.unpack('<Q4s4s4s4s4s4s', byte_string)
+        nanoseconds = (parts[0])
+        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 = 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[int(sixth_entry_bit_string[7:12], 2)]
+        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)
+        # 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)
+
+        return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit,
+                   signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
+                   pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
+                   sector, polarity, df_quality, elevation, azimuth, channel)
+
+    def __init__(self, time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse,
+                 level_unit, signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
+                 pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
+                 sector, polarity, df_quality, elevation, azimuth, channel):
+        r"""
+        I return an instance of an Pulse Data word.
+
+        :param time_of_arrival: nanoseconds since 1970-01-01 00:00:00
+        :type time_of_arrival: Integer
+        :param pdw_format: format code
+        :type pdw_format: Integer
+        :param center_frequency: center frequency in KHz
+        :type center_frequency: Integer
+        :param is_valid: flag to mark if pdw data body is valid
+        :type is_valid: Boolean
+        :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
+        :type level_unit: Integer
+        :param signal_start_missing: signal started before time of arrival
+        :type signal_start_missing: Boolean
+        :param signal_end_missing: signal stops after time of arrival
+        :type signal_end_missing: Boolean
+        :param pulse_width: pulse width in nanoseconds - Zero if no valid pulse detected
+        :type pulse_width: Integer
+        :param frequency_shift_or_bandwidth: Value in KHz - Value set to 1048575 means Unknown
+        :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
+        :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 modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.)
+        :type modulation: String
+        :param sector: reference antenna sector (0-15)
+        :type sector: Integer
+        :param polarity: Horizontal, Vertical, Clockwise, Counter clockwise
+        :type polarity: String
+        :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 channel: detecting channel (0-16) - Zero means unknown
+        :type channel: Integer
+        :return: An instance of class PDW with attributes set according to the data of a data body
+        :rtype: PDW
+        """
+        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 #
+
+    def __str__(self):
+        """
+        I return the string representation of myself.
+
+        :rtype: str
+        """
+        output = ("Time of arrival: " + str(self.time_of_arrival) + "\n" +
+                  "PDW Format identifier: " + str(self.pdw_format_identifier) + "\n" +
+                  "Center frequency: " + str(self.center_frequency) + " KHz\n")
+
+        if self.__is_valid:
+            output += "Signal: Valid\n"
+        else:
+            output += "Signal: Invalid\n"
+
+        if self.is_pulse:
+            output += "Signal type: Pulse\n"
+        else:
+            output += "Signal type: Continuous wave\n"
+
+        if self.level_unit == 1:
+            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 width: " + str(self.pulse_width) + " nanoseconds\n" +
+                   "Frequency shift or bandwidth: " + str(self.frequency_shift_or_bandwidth) + " KHz\n")
+
+        if self.region_of_interest:
+            output += "Region of interest: Yes\n"
+        else:
+            output += "Region of interest: No\n"
+
+        if self.azimuth_confidence == 6.3:
+            output += "Azimuth confidence: Invalid\n"
+        else:
+            output += "Azimuth confidence: " + str(self.azimuth_confidence) + " degree\n"
+
+        output += "Modulation: " + str(self.modulation) + "\n"
+
+        if self.sector == 0:
+            output += "Sector: Unknown\n"
+        else:
+            output += "Sector:" + str(self.sector) + "\n"
+
+        output += "Polarity: " + str(self.polarity) + "\n"
+
+        output += "DF quality: " + str(self.df_quality) + " %\n"
+
+        if self.elevation == 1024:
+            output += "Elevation: Unknown\n"
+        else:
+            output += "Elevation: " + str(self.elevation) + " degree\n"
+
+        if self.azimuth == 409.5:
+            output += "Azimuth: Unknown\n"
+        else:
+            output += "Azimuth: " + str(self.azimuth) + " degree\n"
+
+        output += "Channel: " + str(self.channel) + "\n"
+
+        return output
+
+    def is_valid(self):
+        return self.__is_valid
+
+    def to_json(self):
+        return {'time of arrival': self.time_of_arrival,
+                'FORMATIDENTIFIER': self.pdw_format_identifier,
+                'center frequency': self.center_frequency,
+                'VALID': self.__is_valid,
+                'PULSE': self.is_pulse,
+                'PULSELEVEL': self.pulse_level_or_pulse_field_strength,
+                'PULSEWIDTH': self.pulse_width,
+                'BANDWIDTH': self.frequency_shift_or_bandwidth,
+                'REGIONOFINTEREST': self.region_of_interest,
+                'AZIMUTHCONFIDENCE': self.azimuth_confidence,
+                'MODULATION': self.modulation,
+                'SECTOR': self.sector,
+                'POLARITY': self.polarity,
+                'DFQUALITY': self.df_quality,
+                'ELEVATION': self.elevation,
+                'AZIMUTH': self.azimuth,
+                'CHANNEL': self.channel
+                }
+
+if __name__ == '__main__':
+    pass
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class PDW +(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit, signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth, pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation, sector, polarity, df_quality, elevation, azimuth, channel) +
+
+

I store information from a single ppdw data block.

+
+

Automethod: init

+
+

I return an instance of an Pulse Data word.

+

:param time_of_arrival: nanoseconds since 1970-01-01 00:00:00 +:type time_of_arrival: Integer +:param pdw_format: format code +:type pdw_format: Integer +:param center_frequency: center frequency in KHz +:type center_frequency: Integer +:param is_valid: flag to mark if pdw data body is valid +:type is_valid: Boolean +: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 +:type level_unit: Integer +:param signal_start_missing: signal started before time of arrival +:type signal_start_missing: Boolean +:param signal_end_missing: signal stops after time of arrival +:type signal_end_missing: Boolean +:param pulse_width: pulse width in nanoseconds - Zero if no valid pulse detected +:type pulse_width: Integer +:param frequency_shift_or_bandwidth: Value in KHz - Value set to 1048575 means Unknown +: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 +: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 modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.) +:type modulation: String +:param sector: reference antenna sector (0-15) +:type sector: Integer +:param polarity: Horizontal, Vertical, Clockwise, Counter clockwise +:type polarity: String +: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 channel: detecting channel (0-16) - Zero means unknown +:type channel: Integer +:return: An instance of class PDW with attributes set according to the data of a data body +:rtype: PDW

+
+ +Expand source code + +
class PDW():
+    """
+    I store information from a single ppdw data block.
+
+    .. automethod:: __init__
+    """
+
+    @classmethod
+    def from_bytes(cls, byte_string):
+        """
+        I create an instance of class PDW from data body (8 * 32 bits).
+
+        :param byte_string: a byte string containing a single data body read from a ppdw file
+        :type byte_string: byte string
+
+        :return: an instance of class PDW with attributes set according to the data of a data body
+        :rtype: PDW
+        """
+        assert(len(byte_string) == 32)
+
+        parts = struct.unpack('<Q4s4s4s4s4s4s', byte_string)
+        nanoseconds = (parts[0])
+        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 = 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[int(sixth_entry_bit_string[7:12], 2)]
+        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)
+        # 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)
+
+        return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit,
+                   signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
+                   pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
+                   sector, polarity, df_quality, elevation, azimuth, channel)
+
+    def __init__(self, time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse,
+                 level_unit, signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
+                 pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
+                 sector, polarity, df_quality, elevation, azimuth, channel):
+        r"""
+        I return an instance of an Pulse Data word.
+
+        :param time_of_arrival: nanoseconds since 1970-01-01 00:00:00
+        :type time_of_arrival: Integer
+        :param pdw_format: format code
+        :type pdw_format: Integer
+        :param center_frequency: center frequency in KHz
+        :type center_frequency: Integer
+        :param is_valid: flag to mark if pdw data body is valid
+        :type is_valid: Boolean
+        :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
+        :type level_unit: Integer
+        :param signal_start_missing: signal started before time of arrival
+        :type signal_start_missing: Boolean
+        :param signal_end_missing: signal stops after time of arrival
+        :type signal_end_missing: Boolean
+        :param pulse_width: pulse width in nanoseconds - Zero if no valid pulse detected
+        :type pulse_width: Integer
+        :param frequency_shift_or_bandwidth: Value in KHz - Value set to 1048575 means Unknown
+        :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
+        :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 modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.)
+        :type modulation: String
+        :param sector: reference antenna sector (0-15)
+        :type sector: Integer
+        :param polarity: Horizontal, Vertical, Clockwise, Counter clockwise
+        :type polarity: String
+        :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 channel: detecting channel (0-16) - Zero means unknown
+        :type channel: Integer
+        :return: An instance of class PDW with attributes set according to the data of a data body
+        :rtype: PDW
+        """
+        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 #
+
+    def __str__(self):
+        """
+        I return the string representation of myself.
+
+        :rtype: str
+        """
+        output = ("Time of arrival: " + str(self.time_of_arrival) + "\n" +
+                  "PDW Format identifier: " + str(self.pdw_format_identifier) + "\n" +
+                  "Center frequency: " + str(self.center_frequency) + " KHz\n")
+
+        if self.__is_valid:
+            output += "Signal: Valid\n"
+        else:
+            output += "Signal: Invalid\n"
+
+        if self.is_pulse:
+            output += "Signal type: Pulse\n"
+        else:
+            output += "Signal type: Continuous wave\n"
+
+        if self.level_unit == 1:
+            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 width: " + str(self.pulse_width) + " nanoseconds\n" +
+                   "Frequency shift or bandwidth: " + str(self.frequency_shift_or_bandwidth) + " KHz\n")
+
+        if self.region_of_interest:
+            output += "Region of interest: Yes\n"
+        else:
+            output += "Region of interest: No\n"
+
+        if self.azimuth_confidence == 6.3:
+            output += "Azimuth confidence: Invalid\n"
+        else:
+            output += "Azimuth confidence: " + str(self.azimuth_confidence) + " degree\n"
+
+        output += "Modulation: " + str(self.modulation) + "\n"
+
+        if self.sector == 0:
+            output += "Sector: Unknown\n"
+        else:
+            output += "Sector:" + str(self.sector) + "\n"
+
+        output += "Polarity: " + str(self.polarity) + "\n"
+
+        output += "DF quality: " + str(self.df_quality) + " %\n"
+
+        if self.elevation == 1024:
+            output += "Elevation: Unknown\n"
+        else:
+            output += "Elevation: " + str(self.elevation) + " degree\n"
+
+        if self.azimuth == 409.5:
+            output += "Azimuth: Unknown\n"
+        else:
+            output += "Azimuth: " + str(self.azimuth) + " degree\n"
+
+        output += "Channel: " + str(self.channel) + "\n"
+
+        return output
+
+    def is_valid(self):
+        return self.__is_valid
+
+    def to_json(self):
+        return {'time of arrival': self.time_of_arrival,
+                'FORMATIDENTIFIER': self.pdw_format_identifier,
+                'center frequency': self.center_frequency,
+                'VALID': self.__is_valid,
+                'PULSE': self.is_pulse,
+                'PULSELEVEL': self.pulse_level_or_pulse_field_strength,
+                'PULSEWIDTH': self.pulse_width,
+                'BANDWIDTH': self.frequency_shift_or_bandwidth,
+                'REGIONOFINTEREST': self.region_of_interest,
+                'AZIMUTHCONFIDENCE': self.azimuth_confidence,
+                'MODULATION': self.modulation,
+                'SECTOR': self.sector,
+                'POLARITY': self.polarity,
+                'DFQUALITY': self.df_quality,
+                'ELEVATION': self.elevation,
+                'AZIMUTH': self.azimuth,
+                'CHANNEL': self.channel
+                }
+
+

Static methods

+
+
+def from_bytes(byte_string) +
+
+

I create an instance of class PDW from data body (8 * 32 bits).

+

:param byte_string: a byte string containing a single data body read from a ppdw file +:type byte_string: byte string

+

:return: an instance of class PDW with attributes set according to the data of a data body +:rtype: PDW

+
+ +Expand source code + +
@classmethod
+def from_bytes(cls, byte_string):
+    """
+    I create an instance of class PDW from data body (8 * 32 bits).
+
+    :param byte_string: a byte string containing a single data body read from a ppdw file
+    :type byte_string: byte string
+
+    :return: an instance of class PDW with attributes set according to the data of a data body
+    :rtype: PDW
+    """
+    assert(len(byte_string) == 32)
+
+    parts = struct.unpack('<Q4s4s4s4s4s4s', byte_string)
+    nanoseconds = (parts[0])
+    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 = 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[int(sixth_entry_bit_string[7:12], 2)]
+    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)
+    # 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)
+
+    return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit,
+               signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
+               pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
+               sector, polarity, df_quality, elevation, azimuth, channel)
+
+
+
+

Methods

+
+
+def is_valid(self) +
+
+
+
+ +Expand source code + +
def is_valid(self):
+    return self.__is_valid
+
+
+
+def to_json(self) +
+
+
+
+ +Expand source code + +
def to_json(self):
+    return {'time of arrival': self.time_of_arrival,
+            'FORMATIDENTIFIER': self.pdw_format_identifier,
+            'center frequency': self.center_frequency,
+            'VALID': self.__is_valid,
+            'PULSE': self.is_pulse,
+            'PULSELEVEL': self.pulse_level_or_pulse_field_strength,
+            'PULSEWIDTH': self.pulse_width,
+            'BANDWIDTH': self.frequency_shift_or_bandwidth,
+            'REGIONOFINTEREST': self.region_of_interest,
+            'AZIMUTHCONFIDENCE': self.azimuth_confidence,
+            'MODULATION': self.modulation,
+            'SECTOR': self.sector,
+            'POLARITY': self.polarity,
+            'DFQUALITY': self.df_quality,
+            'ELEVATION': self.elevation,
+            'AZIMUTH': self.azimuth,
+            'CHANNEL': self.channel
+            }
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/PPDWContainer.html =================================================================== --- doc/ammosreader/PPDWContainer.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/PPDWContainer.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,445 @@ + + + + + + +ammosreader.PPDWContainer API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.PPDWContainer

+
+
+
+ +Expand source code + +
import numpy as np
+from datetime import datetime
+
+
+class PPDWContainer():
+    """
+    I store multiple signals imported from one or more .ppdw files
+    .. automethod:: __init__
+    """
+
+    def __init__(self, name, signals=None):
+
+        if signals is None:
+            self.signals = []
+        else:
+            self.signals = signals
+
+        self.name = name
+
+    def __str__(self):
+        return "\n".join(["Number of pulses:" + str(len(self.signals)),
+                          "Number of valid pulses:" + str(self.number_of_valid_pulses()),
+                          "Number of invalid pulses:" + str(self.number_of_invalid_pulses()),
+                          "Start time:" + str(self.start_time()),
+                          "End time:" + str(self.end_time())])
+
+    def add(self, a_pdw):
+        # print(a_pdw)
+        self.signals.append(a_pdw)
+
+    def as_pulse_dict(self):
+        pulse_dict = {}
+        pulse_dict[0] = self.signals[0].to_json()
+        for index, current_pdw in enumerate(self.signals[1:]):
+            pulse_dict[index+1] = current_pdw.to_json()
+            pulse_dict[index]["dtoa"] = (current_pdw.time_of_arrival - pulse_dict[index]["time of arrival"]).item()/1000.0
+        pulse_dict[index+1]["dtoa"] = 0.0  # np.timedelta64(0, 'us')
+        return pulse_dict
+
+    def number_of_valid_pulses(self):
+        return sum(each.is_valid() for each in self.signals)
+
+    def number_of_invalid_pulses(self):
+        return sum(not each.is_valid() for each in self.signals)
+
+    def without_invalids(self):
+        return PPDWContainer(self.name, [each for each in self.signals if each.is_valid()])
+
+    def julian_date_string(self):
+        ts = (self.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's')
+        time_tuple = datetime.utcfromtimestamp(ts).timetuple()
+        return str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3)
+
+    def center_frequencies(self):
+        return list({each.center_frequency for each in self.signals})
+
+    def channels(self):
+        return list({each.channel for each in self.signals})
+
+    def modulations(self):
+        return list({each.modulation for each in self.signals})
+
+    def bandwidths(self):
+        return list({each.frequency_shift_or_bandwidth for each in self.signals})
+
+    def start_time(self):
+        return min([each.time_of_arrival for each in self.signals])
+
+    def end_time(self):
+        return max([each.time_of_arrival for each in self.signals])
+
+    def to_json(self):
+        return {'JULIANDATE': self.julian_date_string(),
+                'STARTTIME': str(self.start_time()),
+                'ENDTIME': str(self.end_time()),
+                'CENTERFREQUENCIES': self.center_frequencies(),
+                'CHANNELS': self.channels(),
+                'MODULATIONS': self.modulations(),
+                'BANDWIDTHS': self.bandwidths()
+                }
+
+if __name__ == '__main__':
+    pass
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class PPDWContainer +(name, signals=None) +
+
+

I store multiple signals imported from one or more .ppdw files

+
+

Automethod: init

+
+
+ +Expand source code + +
class PPDWContainer():
+    """
+    I store multiple signals imported from one or more .ppdw files
+    .. automethod:: __init__
+    """
+
+    def __init__(self, name, signals=None):
+
+        if signals is None:
+            self.signals = []
+        else:
+            self.signals = signals
+
+        self.name = name
+
+    def __str__(self):
+        return "\n".join(["Number of pulses:" + str(len(self.signals)),
+                          "Number of valid pulses:" + str(self.number_of_valid_pulses()),
+                          "Number of invalid pulses:" + str(self.number_of_invalid_pulses()),
+                          "Start time:" + str(self.start_time()),
+                          "End time:" + str(self.end_time())])
+
+    def add(self, a_pdw):
+        # print(a_pdw)
+        self.signals.append(a_pdw)
+
+    def as_pulse_dict(self):
+        pulse_dict = {}
+        pulse_dict[0] = self.signals[0].to_json()
+        for index, current_pdw in enumerate(self.signals[1:]):
+            pulse_dict[index+1] = current_pdw.to_json()
+            pulse_dict[index]["dtoa"] = (current_pdw.time_of_arrival - pulse_dict[index]["time of arrival"]).item()/1000.0
+        pulse_dict[index+1]["dtoa"] = 0.0  # np.timedelta64(0, 'us')
+        return pulse_dict
+
+    def number_of_valid_pulses(self):
+        return sum(each.is_valid() for each in self.signals)
+
+    def number_of_invalid_pulses(self):
+        return sum(not each.is_valid() for each in self.signals)
+
+    def without_invalids(self):
+        return PPDWContainer(self.name, [each for each in self.signals if each.is_valid()])
+
+    def julian_date_string(self):
+        ts = (self.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's')
+        time_tuple = datetime.utcfromtimestamp(ts).timetuple()
+        return str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3)
+
+    def center_frequencies(self):
+        return list({each.center_frequency for each in self.signals})
+
+    def channels(self):
+        return list({each.channel for each in self.signals})
+
+    def modulations(self):
+        return list({each.modulation for each in self.signals})
+
+    def bandwidths(self):
+        return list({each.frequency_shift_or_bandwidth for each in self.signals})
+
+    def start_time(self):
+        return min([each.time_of_arrival for each in self.signals])
+
+    def end_time(self):
+        return max([each.time_of_arrival for each in self.signals])
+
+    def to_json(self):
+        return {'JULIANDATE': self.julian_date_string(),
+                'STARTTIME': str(self.start_time()),
+                'ENDTIME': str(self.end_time()),
+                'CENTERFREQUENCIES': self.center_frequencies(),
+                'CHANNELS': self.channels(),
+                'MODULATIONS': self.modulations(),
+                'BANDWIDTHS': self.bandwidths()
+                }
+
+

Methods

+
+
+def add(self, a_pdw) +
+
+
+
+ +Expand source code + +
def add(self, a_pdw):
+    # print(a_pdw)
+    self.signals.append(a_pdw)
+
+
+
+def as_pulse_dict(self) +
+
+
+
+ +Expand source code + +
def as_pulse_dict(self):
+    pulse_dict = {}
+    pulse_dict[0] = self.signals[0].to_json()
+    for index, current_pdw in enumerate(self.signals[1:]):
+        pulse_dict[index+1] = current_pdw.to_json()
+        pulse_dict[index]["dtoa"] = (current_pdw.time_of_arrival - pulse_dict[index]["time of arrival"]).item()/1000.0
+    pulse_dict[index+1]["dtoa"] = 0.0  # np.timedelta64(0, 'us')
+    return pulse_dict
+
+
+
+def bandwidths(self) +
+
+
+
+ +Expand source code + +
def bandwidths(self):
+    return list({each.frequency_shift_or_bandwidth for each in self.signals})
+
+
+
+def center_frequencies(self) +
+
+
+
+ +Expand source code + +
def center_frequencies(self):
+    return list({each.center_frequency for each in self.signals})
+
+
+
+def channels(self) +
+
+
+
+ +Expand source code + +
def channels(self):
+    return list({each.channel for each in self.signals})
+
+
+
+def end_time(self) +
+
+
+
+ +Expand source code + +
def end_time(self):
+    return max([each.time_of_arrival for each in self.signals])
+
+
+
+def julian_date_string(self) +
+
+
+
+ +Expand source code + +
def julian_date_string(self):
+    ts = (self.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's')
+    time_tuple = datetime.utcfromtimestamp(ts).timetuple()
+    return str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3)
+
+
+
+def modulations(self) +
+
+
+
+ +Expand source code + +
def modulations(self):
+    return list({each.modulation for each in self.signals})
+
+
+
+def number_of_invalid_pulses(self) +
+
+
+
+ +Expand source code + +
def number_of_invalid_pulses(self):
+    return sum(not each.is_valid() for each in self.signals)
+
+
+
+def number_of_valid_pulses(self) +
+
+
+
+ +Expand source code + +
def number_of_valid_pulses(self):
+    return sum(each.is_valid() for each in self.signals)
+
+
+
+def start_time(self) +
+
+
+
+ +Expand source code + +
def start_time(self):
+    return min([each.time_of_arrival for each in self.signals])
+
+
+
+def to_json(self) +
+
+
+
+ +Expand source code + +
def to_json(self):
+    return {'JULIANDATE': self.julian_date_string(),
+            'STARTTIME': str(self.start_time()),
+            'ENDTIME': str(self.end_time()),
+            'CENTERFREQUENCIES': self.center_frequencies(),
+            'CHANNELS': self.channels(),
+            'MODULATIONS': self.modulations(),
+            'BANDWIDTHS': self.bandwidths()
+            }
+
+
+
+def without_invalids(self) +
+
+
+
+ +Expand source code + +
def without_invalids(self):
+    return PPDWContainer(self.name, [each for each in self.signals if each.is_valid()])
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/PPDWReader.html =================================================================== --- doc/ammosreader/PPDWReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/PPDWReader.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,170 @@ + + + + + + +ammosreader.PPDWReader API documentation + + + + + + + + + + + +
+
+
+

Module ammosreader.PPDWReader

+
+
+
+ +Expand source code + +
from pathlib import Path
+
+from ammosreader.PDW import PDW
+from ammosreader.PPDWContainer import PPDWContainer
+
+class PPDWReader():
+
+    def __init__(self, a_file):
+
+        self.file = a_file
+        assert self.file.is_file()
+        self.content = self.file.read_bytes()
+        self.cursor = 0
+        self.container = PPDWContainer(self.file.stem)
+
+    def read_all_frames_left(self):
+        try:
+            while self.cursor <= len(self.content) - 32:
+                current_bytes = self.content[self.cursor:self.cursor+32]
+                assert len(current_bytes) == 32
+                if not current_bytes:
+                    # print('End of file detected')
+                    break
+                if self.cursor + 32 >= len(self.content):
+                    # print('Can not read all 32 bytes of next PDW. EOF')
+                    break
+                self.container.add(PDW.from_bytes(current_bytes))
+                self.cursor += 32
+        except Exception:
+            return None
+        return self.container
+
+
+
+
+
+
+
+
+
+

Classes

+
+
+class PPDWReader +(a_file) +
+
+
+
+ +Expand source code + +
class PPDWReader():
+
+    def __init__(self, a_file):
+
+        self.file = a_file
+        assert self.file.is_file()
+        self.content = self.file.read_bytes()
+        self.cursor = 0
+        self.container = PPDWContainer(self.file.stem)
+
+    def read_all_frames_left(self):
+        try:
+            while self.cursor <= len(self.content) - 32:
+                current_bytes = self.content[self.cursor:self.cursor+32]
+                assert len(current_bytes) == 32
+                if not current_bytes:
+                    # print('End of file detected')
+                    break
+                if self.cursor + 32 >= len(self.content):
+                    # print('Can not read all 32 bytes of next PDW. EOF')
+                    break
+                self.container.add(PDW.from_bytes(current_bytes))
+                self.cursor += 32
+        except Exception:
+            return None
+        return self.container
+
+

Methods

+
+
+def read_all_frames_left(self) +
+
+
+
+ +Expand source code + +
def read_all_frames_left(self):
+    try:
+        while self.cursor <= len(self.content) - 32:
+            current_bytes = self.content[self.cursor:self.cursor+32]
+            assert len(current_bytes) == 32
+            if not current_bytes:
+                # print('End of file detected')
+                break
+            if self.cursor + 32 >= len(self.content):
+                # print('Can not read all 32 bytes of next PDW. EOF')
+                break
+            self.container.add(PDW.from_bytes(current_bytes))
+            self.cursor += 32
+    except Exception:
+        return None
+    return self.container
+
+
+
+
+
+
+
+ +
+ + + Index: doc/ammosreader/index.html =================================================================== --- doc/ammosreader/index.html (revision dff8988960eb5307013e281ff749d9fa74557406) +++ doc/ammosreader/index.html (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,160 @@ + + + + + + +ammosreader API documentation + + + + + + + + + + + +
+
+
+

Package ammosreader

+
+
+
+
+

Sub-modules

+
+
ammosreader.AbstractAmmosReader
+
+

I provide a base class for specialized AmmosReaders.

+
+
ammosreader.AmmosAudioDataBody
+
+

I provide an AMMOS data block for audio data frames.

+
+
ammosreader.AmmosAudioDataHeader
+
+

I provide an AMMOS data header for audio data frames.

+
+
ammosreader.AmmosAudioReader
+
+

I provide a specialized Ammos Reader for audio data.

+
+
ammosreader.AmmosAudioSocketReader
+
+

I read a Ammos datastream from a socket.

+
+
ammosreader.AmmosConstants
+
+

I provide several constants used in R&S software.

+
+
ammosreader.AmmosContainer
+
+

I store the content of an R&S Ammos file in a more accessible way.

+
+
ammosreader.AmmosExtendedAudioDataHeader
+
+

I provide an Ammos extended data header for audio data frames.

+
+
ammosreader.AmmosExtendedIFDataHeader
+
+
+
+
ammosreader.AmmosGlobalFrameBody
+
+

I provide an AMMOS global frame body.

+
+
ammosreader.AmmosGlobalFrameHeader
+
+

I provide an AMMOS global frame header.

+
+
ammosreader.AmmosIFDataBlock
+
+
+
+
ammosreader.AmmosIFDataBlockHeader
+
+
+
+
ammosreader.AmmosIFDataBody
+
+

I provide an AMMOS data body for IF data …

+
+
ammosreader.AmmosIFDataHeader
+
+

I provide a Ammos data header for IF data frames.

+
+
ammosreader.AmmosIFReader
+
+

I provide a specialized Ammos Reader for IF data.

+
+
ammosreader.AmmosSingleFrame
+
+
+
+
ammosreader.IQDWTXBlock
+
+
+
+
ammosreader.PDW
+
+

I store the information of a single PDW block.

+
+
ammosreader.PPDWContainer
+
+
+
+
ammosreader.PPDWReader
+
+
+
+
+
+
+
+
+
+
+
+
+ +
+ + + Index: blic/.buildinfo =================================================================== --- public/.buildinfo (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,4 +1,0 @@ -# Sphinx build info version 1 -# This file hashes the configuration used when building these files. When it is not found, a full rebuild will be done. -config: bd0cf0b1e5e8b74e46ffad0c7feb7650 -tags: 645f666f9bcd5a90fca523b33c5a78b7 Index: pyproject.toml =================================================================== --- pyproject.toml (revision dff8988960eb5307013e281ff749d9fa74557406) +++ pyproject.toml (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,30 @@ +[build-system] +requires = ["poetry-core>=1.0.7"] +build-backend = "poetry.core.masonry.api" + +[tool.poetry] +name = "ammosreader" +version = "0.0.4" +description = "A package to manage ammos datastreams" +authors = ["See Contributors"] +homepage = "https://gitlab.kokyou.de/enno/ammosreader" +repository = "https://gitlab.kokyou.de/enno/ammosreader" +license = "MIT" +readme = "README.md" +packages = [ + { include = "ammosreader" } +] + +include = [ + {path = 'tests/*.py'}, + {path = 'scripts/*.py'}, + {path = 'sampleData/*.csv'}, + {path = 'ammos_logger.conf'} +] +[tool.poetry.scripts] +iqdw_reader = 'scripts.iqdw_reader:main' +pdw_reader = 'scripts.pdw_reader:main' + +[tool.poetry.dependencies] +numpy = "~1.23.2" + Index: qs.txt =================================================================== --- reqs.txt (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,5 +1,0 @@ -numpy==1.21.0 -pymongo==4.1.1 -scipy==1.8.0 -setuptools==58.1.0 -sphinx_adc_theme==0.1.7 Index: quirements.txt =================================================================== --- requirements.txt (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,7 +1,0 @@ -bitstring==3.1.9 -numpy==1.23.0 -Pillow==9.2.0 -pydub==0.25.1 -pymongo==4.1.1 -scipy==1.8.1 -setuptools==58.1.0 Index: mple_scripts/ammos_viewer.py =================================================================== --- sample_scripts/ammos_viewer.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,109 +1,0 @@ -"""I provide a simple bit viewer specialized for AMMOS files.""" - -import tkinter as tk -from PIL import Image, ImageTk -# from random import randbytes -import bitstring - -from ammosreader.AmmosAudioReader import AmmosAudioReader - - -class MainApplication(tk.Frame): - """I implement a simple bit viewer specialized for AMMOS files.""" - - def __init__(self, parent): - """I create the application window.""" - super().__init__(parent) - - self.parent = parent - self.file_name = "../sample_data/audio_data_sample.bin" - self.ammos_container = AmmosAudioReader(self.file_name).read_all_frames_left() - self.file = open(self.file_name, 'rb') - self.bytes = self.file.read(-1) - self.cycle = tk.IntVar() - self.cycle.set(self.ammos_container.unique_frame_sizes()[0]) - self.cache = tk.IntVar() - self.cache.set(500) - self.canvas_size = (800, 600) - self.offset = tk.IntVar() - self.offset.set(0) - self.buildup() - - def buildup(self): - """I build up the initial user interface.""" - self.parent.title("File name:" + self.file_name + " Size:" + str(len(self.bytes))) - self.pack(fill="both", expand=True) - self.default_button_width = 10 - image_frame = tk.Frame(self) - image_frame.pack(side=tk.LEFT) - self.canvas = tk.Canvas(image_frame, bg="#000000", width=self.canvas_size[0], height=self.canvas_size[1]) - self.canvas.pack() - pil_image = Image.frombytes("1", self.canvas_size, self.current_bytes()) - self.image = ImageTk.PhotoImage(pil_image) - # pil_image.show() - button_frame = tk.Frame(self) - self.image_id = self.canvas.create_image(0, 0, anchor='nw', image=self.image) - button_frame.pack(side="right") - - tk.Label(button_frame, text='Cycle in bytes').pack(side=tk.TOP) - self.cycle_entry = tk.Entry(button_frame, text=self.cycle) - self.cycle_entry.pack(side=tk.TOP) - - tk.Label(button_frame, text='Cache in Megabytes').pack(side=tk.TOP) - self.cache_entry = tk.Entry(button_frame, text=self.cache) - self.cache_entry.pack(side=tk.TOP) - - tk.Label(button_frame, text='Offset in bytes').pack(side=tk.TOP) - self.offset_entry = tk.Entry(button_frame, text=self.offset) - self.offset_entry.pack(side=tk.TOP) - - tk.Button(button_frame, - text="Update", - width=self.default_button_width, - command=self.update_canvas).pack(side=tk.TOP, fill=tk.BOTH, expand=True) - - tk.Button( - button_frame, - text="Load file", - width=self.default_button_width, - command=self.load_file).pack(side="top", fill="both", expand=True) - - tk.Button( - button_frame, - text="Quit", - width=self.default_button_width, - command=self.quit).pack(side="top", fill="both", expand=True) - - def current_bytes(self): - """I return the current bytes to display in the canvas.""" - if self.offset.get() % 8 == 0: - return b"".join([self.bytes[each: each+self.canvas_size[0]] for each in range(self.offset.get(), - len(self.bytes), - self.cycle.get())]) - bits = bitstring.BitArray(self.bytes) - new_bytes = bits[self.offset.get()*8:].tobytes() - return b"".join([new_bytes[each: each+self.canvas_size[0]] for each in range(0, - len(new_bytes), - self.cycle.get())]) - - def update_canvas(self): - """I update the canvas.""" - self.image = ImageTk.PhotoImage(Image.frombytes("L", self.canvas_size, self.current_bytes())) - self.canvas.itemconfig(self.image_id, image=self.image) - - def load_file(self): - """I load a binary ammos file.""" - pass - - def quit(self): - """I quit the application.""" - self.parent.destroy() - - -if __name__ == '__main__': - root = tk.Tk() - root.geometry("1920x1200") - - app = MainApplication(root) - - root.mainloop() Index: mple_scripts/audio_reader.py =================================================================== --- sample_scripts/audio_reader.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,34 +1,0 @@ -import sys -import io -from pydub import AudioSegment -from pydub.playback import play - -from ammosreader.AmmosAudioReader import AmmosAudioReader - -if __name__ == '__main__': - - if len(sys.argv) != 2: - sys.exit() - - file_name = sys.argv[1] - - print("File name:", file_name) - - dat_file = AmmosAudioReader(file_name) - - dat_file.read_all_frames_left() - print("Sample rate:", dat_file.container.global_frames[0].global_frame_body.data_header.sample_rate) - print("Container size:", dat_file.container.size()) - print("Frequencies", dat_file.container.frequencies()) - print("Frame types:", dat_file.container.frame_types()) - print("Unique frame sizes", dat_file.container.unique_frame_sizes()) - print("Total frame size:", sum(dat_file.container.frame_sizes())) - print("Homogenic:", dat_file.container.is_homogenic()) - pcm_data = dat_file.pcm_for_channel(0) - print("PCM data size total:", len(pcm_data)) - data = AudioSegment.from_raw(io.BytesIO(pcm_data), - sample_width=2, - frame_rate=22050, - channels=1) - print("Start playing audio") - play(data) Index: mple_scripts/audio_socket_reader_test.py =================================================================== --- sample_scripts/audio_socket_reader_test.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,41 +1,0 @@ -import sys -import os - -sys.path.append('../src/') - -import socket - -import numpy as np -import scipy.io.wavfile as wavfile - -from ammosreader.AmmosAudioSocketReader import AmmosAudioSocketReader - - -if __name__ == '__main__': - frames = [] - - # create socket - in_socket = socket.socket() - in_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - in_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) - in_socket.bind(('127.0.0.1', 12345)) - print('Waiting for connection') - in_socket.listen(1) - conn, _ = in_socket.accept() - print('Connected') - - # create AmmosAudioSocketReader - socket_reader = AmmosAudioSocketReader(conn) - - print('Reading Frames') - while True: - try: - frames.append(socket_reader.read_next_frame()) - except KeyboardInterrupt: - break - - # write frames to wav file - print('Writing frames to Wav File') - audio = np.concatenate([frame[0] for frame in frames]) - print(audio) - wavfile.write(f"test.wav", frames[0][1], audio) Index: mple_scripts/audio_socket_reader_test_streamer.py =================================================================== --- sample_scripts/audio_socket_reader_test_streamer.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,28 +1,0 @@ -import sys -import os -sys.path.append('../src/') -import socket - -try: - file_path = sys.argv[1] -except IndexError: - print("Please provide a file path as argument") - sys.exit(1) -except FileNotFoundError: - print("File not found") - sys.exit(1) - -file = open(file_path, 'rb') -total_bytes = file.read() -print('File read') - -print('Connecting to socket') -out_socket = socket.socket() -out_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) -out_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) -out_socket.connect(('127.0.0.1', 12345)) -print('Connected') -print('Sending Bytes') -out_socket.sendall(total_bytes) -out_socket.close() -out_socket.shutdown(socket.SHUT_RDWR) Index: mple_scripts/import_signal.py =================================================================== --- sample_scripts/import_signal.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,94 +1,0 @@ -import sys -import re -import sqlite3 -import argparse -import numpy as np -from datetime import datetime -from pathlib import Path -from pymongo import MongoClient -from ammosreader.AmmosIFReader import AmmosIFReader -from ammosreader.PPDWReader import PPDWReader - -parser = argparse.ArgumentParser() -parser.add_argument('-s', '--source', required=True, help='specify source for given signal') -parser.add_argument('-i', '--input-dir', required=True, help='specify source directory') -parser.add_argument('-o', '--output-file', required=True, help='specify sqlite3 database file') -parser.add_argument('-n', '--signal-number', required=True, help='specify number for signal') -parser.add_argument('-c', '--code', required=True, help='specify code for signal') - -args = parser.parse_args() - -source = args.source.upper() -code = args.code.upper() -signal_number = args.signal_number -input_dir = Path(args.input_dir) -sqlite3_file = Path(args.output_file) - -if not re.match('[A-Z0-9]{2}', source): - print("Source identifier", source, "invalid") - sys.exit() - -if not re.match('[A-Z][0-9]{3}[A-Z]', code): - print("ELINT-Code incorrect") - sys.exit(1) - -if not re.match('[0-9]{4}', signal_number): - print("Signal number incorrect") - sys.exit(1) - -if not input_dir.is_dir(): - print("Input dir invalid") - sys.exit(1) - -dir_name = input_dir.name - -files_inside = [each for each in (input_dir.iterdir()) if (each.suffix in ['.idx', '.iqdw', '.ppdw'] and each.stem == dir_name)] - -if len(files_inside) != 3: - print(".idx, .iqdw or .ppdw file missing in", input_dir) - sys.exit(1) - -iqdw_file = (input_dir / input_dir.stem).with_suffix('.iqdw') -ppdw_file = (input_dir / input_dir.stem).with_suffix('.ppdw') -idx_file = (input_dir / input_dir.stem).with_suffix('.idx') - -#if not sqlite3_file.is_file(): -# print("Sqlite3 database file not found at", sqlite3_file) -# sys.exit(1) - -# try: -# connection = sqlite3.connect(str(sqlite3_file)) -# cursor = connection.cursor() -# except Exception: -# print("Can not connect to database file", str(sqlite3_file)) - # sys.exit(1) - -#ammos_if_reader = AmmosIFReader(str(iqdw_file)) -#ammos_if_reader.read_all_frames_left() -# for each_frame in ammos_if_reader.container.global_frames: -# print(each_frame) - -ppdw_reader = PPDWReader(ppdw_file) -ppdw_reader.read_all_frames_left() - -client = MongoClient('hackathon.kid.local', 27017) - -database = client['ELINT'] -ts = (ppdw_reader.container.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's') -time_tuple = datetime.utcfromtimestamp(ts).timetuple() -julian_date_string = str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3) -code_collection = database[code] -year_collection = code_collection[str(time_tuple.tm_year)] -source_collection = year_collection[source] - -json_representation = ppdw_reader.container.to_json() -json_representation['SIGNALNUMBER'] = signal_number -json_representation['SOURCEFILE'] = str(ppdw_file) -source_collection.insert_one(json_representation) - -# for each in ppdw_reader.container.signals: -# each_json = each.to_json() -# each_json['SIGNALNUMBER'] = signal_number -# each_json['JULIANDATE'] = julian_date_string -# each_json['SOURCEFILE'] = str(ppdw_file) -# source_collection.insert_one(each_json) Index: mple_scripts/import_signals.py =================================================================== --- sample_scripts/import_signals.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,80 +1,0 @@ -"""I import .ppdw files from multiple subdirectories into a mongodb.""" - -import sys -import re -import argparse -import numpy as np -from datetime import datetime -from pathlib import Path -from pymongo import MongoClient -from ammosreader.PPDWReader import PPDWReader - -parser = argparse.ArgumentParser() -parser.add_argument('-s', '--source', required=True, help='specify source for given signal') -parser.add_argument('-i', '--input-dir', required=True, help='specify source directory') - -args = parser.parse_args() - -source = args.source.upper() -input_dir = Path(args.input_dir) - -if not re.match('[A-Z0-9]{2}', source): - print("Source identifier", source, "invalid") - sys.exit() - -code = input_dir.name - -# FIXME: Directory structure differs -# FIXME: get julian date, source identifier, signal number and ELINT-Code from input_dir path - -# if not re.match('[A-Z][0-9]{3}[A-Z]', code): -# print("ELINT-Code incorrect") -# sys.exit(1) - - -if not input_dir.is_dir(): - print("Input dir invalid") - sys.exit(1) - -subdirs = [f for f in input_dir.iterdir() if f.is_dir()] - -client = MongoClient('hackathon.kid.local', 27017) - -for each_dir in subdirs: - - # print("Each dir", each_dir) - dir_name = each_dir.name - - # if not re.match('[0-9]{5}', dir_name): - # print("Signal number incorrect") - # sys.exit(1) - - files_inside = [each for each in (each_dir.iterdir()) if (each.suffix in ['.idx', '.iqdw', '.ppdw'] and - each_dir.stem == dir_name)] - - # print("Files inside", files_inside) - if len(files_inside) != 3: - print(".idx, .iqdw or .ppdw file missing in", each_dir) - sys.exit(1) - - iqdw_file = (input_dir / each_dir / each_dir.stem).with_suffix('.iqdw') - ppdw_file = (input_dir / each_dir / each_dir.stem).with_suffix('.ppdw') - # print("PPDW file", ppdw_file) - idx_file = (input_dir / each_dir / each_dir.stem).with_suffix('.idx') - - ppdw_reader = PPDWReader(ppdw_file) - ppdw_reader.read_all_frames_left() - - database = client['ELINT'] - ts = (ppdw_reader.container.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's') - time_tuple = datetime.utcfromtimestamp(ts).timetuple() - julian_date_string = str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3) - print("Julian", julian_date_string, str(ppdw_file)) - code_collection = database[input_dir.name] - year_collection = code_collection[str(time_tuple.tm_year)] - source_collection = year_collection[source] - - json_representation = ppdw_reader.container.to_json() - json_representation['SIGNALNUMBER'] = dir_name - json_representation['SOURCEFILE'] = str(ppdw_file) - source_collection.insert_one(json_representation) Index: mple_scripts/iqdw_reader.py =================================================================== --- sample_scripts/iqdw_reader.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,20 +1,0 @@ -import sys - -from ammosreader.AmmosIFReader import AmmosIFReader - -if __name__ == '__main__': - - if len(sys.argv) != 2: - sys.exit() - - file_name = sys.argv[1] - - print("File name:", file_name) - - dat_file = AmmosIFReader(file_name) - - dat_file.read_all_frames_left() - print("Frequencies", dat_file.container.frequencies()) - print("Frame types:", dat_file.container.frame_types()) - print("Total frame sizes:", sum(dat_file.container.frame_sizes())) - print("Homogenic:", dat_file.container.is_homogenic()) Index: mple_scripts/pdw_reader.py =================================================================== --- sample_scripts/pdw_reader.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,27 +1,0 @@ -import os -import sys -sys.path.append('../src/') - -from ammosreader.PDW import PDW - -if __name__ == '__main__': - - if len(sys.argv) != 2: - print("Specify name of .ppdw file") - sys.exit() - - file_name = sys.argv[1] - - with open(file_name, 'rb') as f: - - while(True): - current_bytes = f.read(32) - if current_bytes == '': - print("End of file detected") - break - if len(current_bytes) != 32: - print("Can not read all 32 bytes of next PDW") - break - - current_pdw = PDW.from_bytes(current_bytes) - print(current_pdw) Index: scripts/iqdw_reader.py =================================================================== --- scripts/iqdw_reader.py (revision dff8988960eb5307013e281ff749d9fa74557406) +++ scripts/iqdw_reader.py (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,19 @@ +import sys + +from ammosreader.AmmosIFReader import AmmosIFReader + +def main(): + if len(sys.argv) != 2: + sys.exit() + + file_name = sys.argv[1] + + print("File name:", file_name) + + dat_file = AmmosIFReader(file_name) + + dat_file.read_all_frames_left() + print("Frequencies", dat_file.container.frequencies()) + print("Frame types:", dat_file.container.frame_types()) + print("Total frame sizes:", sum(dat_file.container.frame_sizes())) + print("Homogenic:", dat_file.container.is_homogenic()) Index: scripts/pdw_reader.py =================================================================== --- scripts/pdw_reader.py (revision dff8988960eb5307013e281ff749d9fa74557406) +++ scripts/pdw_reader.py (revision dff8988960eb5307013e281ff749d9fa74557406) @@ -0,0 +1,25 @@ +import sys + +from ammosreader.PDW import PDW + +def main(): + + if len(sys.argv) != 2: + print("Specify name of .ppdw file") + sys.exit() + + file_name = sys.argv[1] + + with open(file_name, 'rb') as f: + + while(True): + current_bytes = f.read(32) + if current_bytes == '': + print("End of file detected") + break + if len(current_bytes) != 32: + print("Can not read all 32 bytes of next PDW") + break + + current_pdw = PDW.from_bytes(current_bytes) + print(current_pdw) Index: tup.py =================================================================== --- setup.py (revision 29bf25ae450e76dd61536c549c5adc13d8a4288f) +++ (revision ) @@ -1,7 +1,0 @@ -from setuptools import setup - -setup( - name="ammosreader", - version="1.2.7", - packages=['ammosreader',], - scripts=['sample_scripts/iqdw_reader.py'])